From 21c9119bcb2ea82a2d2ace8427cbf7dfc4679b01 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Mon, 10 Apr 2023 03:12:39 +0100 Subject: [PATCH] Add management of the retries via headers. --- base.go | 10 ++-- base_test.go | 34 ++++++++++-- query.go | 31 +++++------ query_test.go | 18 ++++++- request.go | 12 ++++- structs.go | 39 +++++++------- utils/concurrency/concurrency.go | 30 ----------- utils/concurrency/concurrency_test.go | 74 --------------------------- 8 files changed, 95 insertions(+), 153 deletions(-) delete mode 100644 utils/concurrency/concurrency.go delete mode 100644 utils/concurrency/concurrency_test.go diff --git a/base.go b/base.go index 2c18f19..88dbf52 100644 --- a/base.go +++ b/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" @@ -22,11 +21,10 @@ import ( func NewConnection() *BaseClient { b := &BaseClient{ - client: &http.Client{}, - concurrencyManager: concurrency.NewPool(-1), - MaxGoRoutines: -1, - LoggingLevel: logger.Warn, - LoggerColorful: true, + client: &http.Client{}, + MaxGoRoutines: -1, + LoggingLevel: logger.Warn, + LoggerColorful: true, } if b.LoggerWriter == nil { diff --git a/base_test.go b/base_test.go index 4ab7fad..788f77b 100644 --- a/base_test.go +++ b/base_test.go @@ -10,16 +10,32 @@ import ( func TestNewConnection(t *testing.T) { tests := []struct { - want *BaseClient - name string + 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) }) } } diff --git a/query.go b/query.go index 62a8e83..eca5f11 100644 --- a/query.go +++ b/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 @@ -92,12 +94,13 @@ func (c *BaseClient) Query(queryContent string, queryVariables interface{}, quer } q := &queryExecutor{ - client: c, - query: compiledQuery.compiledQuery, - headers: parseQueryHeaders, - context: context.Background(), - should_cache: should_cache, - hash: queryHash, + client: c, + query: compiledQuery.compiledQuery, + headers: parseQueryHeaders, + context: context.Background(), + should_cache: should_cache, + retries_enabled: retries_enabled, + hash: queryHash, } q.execute() @@ -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()) - } - if cache_enabled != c.cache.enabled { - headers_modified = true + panic(fmt.Sprintf("Unable to parse gqlcache value %s", err.Error())) } + 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 } diff --git a/query_test.go b/query_test.go index 250c0a8..9808c53 100644 --- a/query_test.go +++ b/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() - if gotReturnHeaders, _, _ := c.parseQueryHeaders(tt.args.queryHeaders); !reflect.DeepEqual(gotReturnHeaders, tt.wantReturnHeaders) { + q := &Query{} + if gotReturnHeaders, _, _, _ := q.parseQueryHeaders(tt.args.queryHeaders); !reflect.DeepEqual(gotReturnHeaders, tt.wantReturnHeaders) { t.Errorf("BaseClient.parseQueryHeaders() = %v, want %v", gotReturnHeaders, tt.wantReturnHeaders) } }) diff --git a/request.go b/request.go index 175ca4b..38b859b 100644 --- a/request.go +++ b/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()) diff --git a/structs.go b/structs.go index bc4a8fc..0b700fa 100644 --- a/structs.go +++ b/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" ) @@ -22,18 +21,17 @@ type retriesConfig struct { } type BaseClient struct { - Logger Logger - LoggerWriter Writer - client *http.Client - concurrencyManager *concurrency.Pool - endpoint string - responseType string - cache cacheStore - MaxGoRoutines int - LoggingLevel logger.LogLevel - LoggerColorful bool - validate bool - retries retriesConfig + Logger Logger + LoggerWriter Writer + client *http.Client + endpoint string + responseType string + cache cacheStore + MaxGoRoutines int + LoggingLevel logger.LogLevel + LoggerColorful bool + validate bool + retries retriesConfig } type Query struct { @@ -44,13 +42,14 @@ type Query struct { } type queryExecutor struct { - result queryExecutorResult - context context.Context - client *BaseClient - headers map[string]interface{} - hash string - query []byte - should_cache bool + result queryExecutorResult + context context.Context + client *BaseClient + headers map[string]interface{} + hash string + query []byte + should_cache bool + retries_enabled bool } type queryExecutorResult struct { diff --git a/utils/concurrency/concurrency.go b/utils/concurrency/concurrency.go deleted file mode 100644 index 5ccb124..0000000 --- a/utils/concurrency/concurrency.go +++ /dev/null @@ -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...) - }() -} diff --git a/utils/concurrency/concurrency_test.go b/utils/concurrency/concurrency_test.go deleted file mode 100644 index e81b7e2..0000000 --- a/utils/concurrency/concurrency_test.go +++ /dev/null @@ -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...) - }) - } -}