Add management of the retries via headers.

base.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/akyoto/cache"
-	"github.com/lukaszraczylo/go-simple-graphql/utils/concurrency"
 	"github.com/lukaszraczylo/go-simple-graphql/utils/logger"
 
 	"golang.org/x/net/http2"
@@ -23,7 +22,6 @@ func NewConnection() *BaseClient {
 
 	b := &BaseClient{
 		client:         &http.Client{},
-		concurrencyManager: concurrency.NewPool(-1),
 		MaxGoRoutines:  -1,
 		LoggingLevel:   logger.Warn,
 		LoggerColorful: true,

base_test.go

@@ -11,15 +11,31 @@ import (
 func TestNewConnection(t *testing.T) {
 	tests := []struct {
 		want     *BaseClient
+		cache    bool
+		endpoint string
+		output   string
 		name     string
 	}{
 		{
-			name: "Test NewConnection",
+			name:     "Test NewConnection https",
+			endpoint: "https://api.github.com/graphql",
+			output:   "string",
+			cache:    true,
 			want: &BaseClient{
 				endpoint:     "https://api.github.com/graphql",
 				responseType: "string",
 			},
 		},
+		{
+			name:     "Test NewConnection http",
+			endpoint: "http://api.github.com/graphql",
+			output:   "mapstring",
+			cache:    false,
+			want: &BaseClient{
+				endpoint:     "http://api.github.com/graphql",
+				responseType: "mapstring",
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -27,11 +43,19 @@ func TestNewConnection(t *testing.T) {
 			// and GRAPHQL_OUTPUT to string
 			current := os.Getenv("GRAPHQL_ENDPOINT")
 			defer os.Setenv("GRAPHQL_ENDPOINT", current)
-			os.Setenv("GRAPHQL_ENDPOINT", "https://api.github.com/graphql")
+			os.Setenv("GRAPHQL_ENDPOINT", tt.endpoint)
-
 			got := NewConnection()
+			got.SetEndpoint(tt.endpoint)
+			got.SetOutput(tt.output)
+			if tt.cache {
+				got.enableCache()
+			} else {
+				got.disableCache()
+			}
+
 			assert.Equal(t, tt.want.endpoint, got.endpoint)
 			assert.Equal(t, tt.want.responseType, got.responseType)
+			assert.Equal(t, tt.cache, got.cache.enabled)
 		})
 	}
 }

query.go

@@ -70,15 +70,17 @@ func (c *BaseClient) NewQuery(q ...any) *Query {
 func (c *BaseClient) Query(queryContent string, queryVariables interface{}, queryHeaders map[string]interface{}) (any, error) {
 
 	compiledQuery := c.NewQuery(queryContent, queryVariables)
-	parseQueryHeaders, enabledCache, headersModified := c.parseQueryHeaders(queryHeaders)
+	parseQueryHeaders, enabledCache, headersModified, enabledRetries := compiledQuery.parseQueryHeaders(queryHeaders)
 
-	var should_cache bool
+	var should_cache, retries_enabled bool
 	var queryHash string
 
 	if headersModified {
 		should_cache = enabledCache
+		retries_enabled = enabledRetries
 	} else {
 		should_cache = c.cache.enabled
+		retries_enabled = c.retries.enabled
 	}
 
 	// if queryContent does not start with `query` then we don't want to cache it
@@ -97,6 +99,7 @@ func (c *BaseClient) Query(queryContent string, queryVariables interface{}, quer
 		headers:         parseQueryHeaders,
 		context:         context.Background(),
 		should_cache:    should_cache,
+		retries_enabled: retries_enabled,
 		hash:            queryHash,
 	}
 
@@ -126,7 +129,7 @@ func (c *BaseClient) decodeResponse(jsonData []byte) any {
 	}
 }
 
-func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (returnHeaders map[string]interface{}, cache_enabled bool, headers_modified bool) {
+func (q *Query) parseQueryHeaders(queryHeaders map[string]interface{}) (returnHeaders map[string]interface{}, cache_enabled bool, headers_modified bool, retries_enabled bool) {
 	returnHeaders = make(map[string]interface{})
 	var err error
 
@@ -134,18 +137,16 @@ func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (ret
 		if k == "gqlcache" {
 			cache_enabled, err = strconv.ParseBool(fmt.Sprintf("%v", v))
 			if err != nil {
-				c.Logger.Error(c, "Can't parse gqlcache header", "error", err.Error())
+				panic(fmt.Sprintf("Unable to parse gqlcache value %s", err.Error()))
 			}
-			if cache_enabled != c.cache.enabled {
 			headers_modified = true
-			}
 			continue
 		}
 		if k == "gqlretries" {
-			c.retries.enabled, _ = strconv.ParseBool(fmt.Sprintf("%v", v))
+			retries_enabled, _ = strconv.ParseBool(fmt.Sprintf("%v", v))
 			continue
 		}
 		returnHeaders[k] = v
 	}
-	return returnHeaders, cache_enabled, headers_modified
+	return returnHeaders, cache_enabled, headers_modified, retries_enabled
 }

query_test.go

@@ -134,6 +134,20 @@ func TestBaseClient_decodeResponse(t *testing.T) {
 			},
 			want: []byte(`{"data":{"hello":"world"}}`),
 		},
+		{
+			name: "Test decodeResponse - mapstring",
+			fields: fields{
+				responseType: "mapstring",
+			},
+			args: args{
+				jsonData: []byte(`{"data":{"hello":"world"}}`),
+			},
+			want: map[string]interface{}{
+				"data": map[string]interface{}{
+					"hello": "world",
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -333,8 +347,8 @@ func TestBaseClient_parseQueryHeaders(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			c := NewConnection()
+			q := &Query{}
-			if gotReturnHeaders, _, _ := c.parseQueryHeaders(tt.args.queryHeaders); !reflect.DeepEqual(gotReturnHeaders, tt.wantReturnHeaders) {
+			if gotReturnHeaders, _, _, _ := q.parseQueryHeaders(tt.args.queryHeaders); !reflect.DeepEqual(gotReturnHeaders, tt.wantReturnHeaders) {
 				t.Errorf("BaseClient.parseQueryHeaders() = %v, want %v", gotReturnHeaders, tt.wantReturnHeaders)
 			}
 		})

request.go

@@ -23,6 +23,11 @@ func (c *BaseClient) executeQuery(query []byte, headers any) (result any, err er
 		httpRequest.Header.Add(key, fmt.Sprintf("%s", value))
 	}
 
+	var retries_available = c.retries.max
+	if !c.retries.enabled {
+		retries_available = 1
+	}
+
 	var httpResponse *http.Response
 
 	err = retry.Do(
@@ -56,7 +61,7 @@ func (c *BaseClient) executeQuery(query []byte, headers any) (result any, err er
 		retry.OnRetry(func(n uint, err error) {
 			c.Logger.Error(c, "Retrying query", "error", err.Error())
 		}),
-		retry.Attempts(uint(c.retries.max)),
+		retry.Attempts(uint(retries_available)),
 		retry.DelayType(retry.BackOffDelay),
 		retry.Delay(time.Duration(c.retries.delay)*time.Second),
 		retry.LastErrorOnly(true),
@@ -80,6 +85,11 @@ func (q *queryExecutor) execute() {
 		}
 	}
 
+	if q.retries_enabled {
+		q.client.Logger.Debug(q.client, "Retries enabled")
+		q.client.retries.enabled = true
+	}
+
 	response, err := q.client.executeQuery(q.query, q.headers)
 	if err != nil {
 		q.client.Logger.Error(q.client, "Error while executing query;", "error", err.Error())

structs.go

@@ -5,7 +5,6 @@ import (
 	"net/http"
 
 	"github.com/akyoto/cache"
-	"github.com/lukaszraczylo/go-simple-graphql/utils/concurrency"
 	"github.com/lukaszraczylo/go-simple-graphql/utils/logger"
 )
 
@@ -25,7 +24,6 @@ type BaseClient struct {
 	Logger         Logger
 	LoggerWriter   Writer
 	client         *http.Client
-	concurrencyManager *concurrency.Pool
 	endpoint       string
 	responseType   string
 	cache          cacheStore
@@ -51,6 +49,7 @@ type queryExecutor struct {
 	hash            string
 	query           []byte
 	should_cache    bool
+	retries_enabled bool
 }
 
 type queryExecutorResult struct {

utils/concurrency/concurrency.go

@@ -1,30 +0,0 @@
-package concurrency
-
-type Pool struct {
-	queue chan struct{}
-	jobs  int
-}
-
-func NewPool(size int) *Pool {
-	if size < 0 {
-		size = 0
-	}
-	return &Pool{
-		jobs:  size,
-		queue: make(chan struct{}, size),
-	}
-}
-
-func (p *Pool) Enqueue(job func(params ...any), params ...any) {
-	if p.jobs == 0 {
-		go job(params...)
-		return
-	}
-	p.queue <- struct{}{}
-	go func() {
-		defer func() {
-			<-p.queue
-		}()
-		job(params...)
-	}()
-}

utils/concurrency/concurrency_test.go

@@ -1,74 +0,0 @@
-package concurrency
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestNewPool(t *testing.T) {
-	type args struct {
-		size int
-	}
-	tests := []struct {
-		want *Pool
-		name string
-		args args
-	}{
-		{
-			name: "TestNewPool",
-			args: args{
-				size: 0,
-			},
-			want: &Pool{
-				jobs:  0,
-				queue: make(chan struct{}, 0),
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := NewPool(tt.args.size)
-			assert.Equal(t, got.jobs, tt.want.jobs)
-			assert.Equal(t, cap(got.queue), cap(tt.want.queue))
-		})
-	}
-}
-
-func TestPool_Enqueue(t *testing.T) {
-	type fields struct {
-		queue chan struct{}
-		jobs  int
-	}
-	type args struct {
-		job    func(params ...any)
-		params []any
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-	}{
-		{
-			name: "TestPool_Enqueue",
-			fields: fields{
-				queue: make(chan struct{}, 0),
-				jobs:  0,
-			},
-			args: args{
-				job: func(params ...any) {
-					return
-				},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			p := &Pool{
-				queue: tt.fields.queue,
-				jobs:  tt.fields.jobs,
-			}
-			p.Enqueue(tt.args.job, tt.args.params...)
-		})
-	}
-}