Memleak fix

README.md

@@ -9,6 +9,7 @@ Ps. It's Hasura friendly.
   - [Features](#features)
   - [Usage example](#usage-example)
     - [Environment variables](#environment-variables)
+    - [Modifiers on the fly](#modifiers-on-the-fly)
     - [Cache](#cache)
     - [Example reader code](#example-reader-code)
     - [Tips](#tips)
@@ -40,6 +41,11 @@ Therefore, I present you the simple client to which you can copy & paste your gr
 * `GRAPHQL_RETRIES_NUMBER` - Number of retries: Default: `1`
 * `GRAPHQL_RETRIES_DELAY` - Delay in retries in milliseconds. Default: `250`
 
+### Modifiers on the fly
+
+* `gql.SetEndpoint('your-endpoint-url')` - modifies endpoint, without the need to set the environment variable
+* `gql.SetOutput('byte')` - modifies output format, without the need to set the environment variable
+
 ### Cache
 
 You have two options to enable the cache:

base.go

@@ -2,8 +2,10 @@ package gql
 
 import (
 	"crypto/tls"
+	"fmt"
 	"time"
 
+	"github.com/allegro/bigcache"
 	"github.com/lukaszraczylo/go-simple-graphql/utils/concurrency"
 	"github.com/lukaszraczylo/go-simple-graphql/utils/logger"
 
@@ -14,7 +16,6 @@ import (
 	"os"
 	"strings"
 
-	"github.com/allegro/bigcache"
 	"github.com/gookit/goutil/envutil"
 )
 
@@ -42,13 +43,7 @@ func NewConnection() *BaseClient {
 	b.retries.max = envutil.GetInt("GRAPHQL_RETRIES_MAX", 1)
 	b.retries.delay = envutil.GetInt("GRAPHQL_RETRIES_DELAY", 300)
 
-	if b.cache.enabled {
-		var err error
-		b.cache.client, err = bigcache.NewBigCache(bigcache.DefaultConfig(time.Duration(b.cache.ttl) * time.Second))
-		if err != nil {
-			b.Logger.Error(b, "Error while creating cache client;", "error", err.Error())
-		}
-	}
+	b.enableCache()
 
 	b_tmp_log_level := envutil.Getenv("LOG_LEVEL", "info")
 
@@ -71,9 +66,20 @@ func NewConnection() *BaseClient {
 	})
 
 	var httpClient *http.Client
+
+	httpTransport := &http.Transport{
+		MaxIdleConns:          10,
+		IdleConnTimeout:       15 * time.Second,
+		ResponseHeaderTimeout: 15 * time.Second,
+		DisableKeepAlives:     false,
+	}
+
 	if strings.HasPrefix(b.endpoint, "http://") {
 		httpClient = &http.Client{
-			Transport: http.DefaultTransport,
+			Transport: httpTransport,
+			CheckRedirect: func(req *http.Request, via []*http.Request) error {
+				return http.ErrUseLastResponse
+			},
 		}
 	} else {
 		tlsClientConfig := &tls.Config{}
@@ -83,6 +89,8 @@ func NewConnection() *BaseClient {
 		http2Transport := &http2.Transport{
 			AllowHTTP:       true,
 			TLSClientConfig: tlsClientConfig,
+			ReadIdleTimeout: 15 * time.Second,
+			PingTimeout:     15 * time.Second,
 		}
 		httpClient = &http.Client{
 			Transport: http2Transport,
@@ -102,3 +110,31 @@ func (b *BaseClient) SetEndpoint(endpoint string) {
 func (b *BaseClient) SetOutput(output string) {
 	b.responseType = output
 }
+
+func (b *BaseClient) enableCache() {
+	var err error
+	b.cache.client, err = bigcache.NewBigCache(
+		// bigcache.DefaultConfig(time.Duration(b.cache.ttl) * time.Second)
+		bigcache.Config{
+			Shards:      1024,
+			LifeWindow:  time.Duration(b.cache.ttl) * time.Second,
+			CleanWindow: time.Duration(b.cache.ttl) * 2 * time.Second,
+			OnRemove: func(key string, entry []byte) {
+				b.Logger.Debug(b, "Removing cache entry;", "key", key)
+			},
+			MaxEntriesInWindow: 1000 * 10 * 60,
+			MaxEntrySize:       500,
+			Verbose:            true,
+			HardMaxCacheSize:   8192,
+		},
+	)
+	if err != nil {
+		fmt.Println(">> Error while creating cache client;", "error", err.Error())
+		panic(err)
+	}
+}
+
+func (b *BaseClient) disableCache() {
+	b.Logger.Debug(b, "Disabling cache")
+	b.cache.enabled = false
+}

query.go

@@ -3,12 +3,10 @@ package gql
 import (
 	"context"
 	"fmt"
-	"reflect"
 	"strconv"
 
-	"github.com/lukaszraczylo/go-simple-graphql/utils/helpers"
-
 	"github.com/gookit/goutil/strutil"
+	"github.com/lukaszraczylo/go-simple-graphql/utils/helpers"
 
 	jsoniter "github.com/json-iterator/go"
 )
@@ -70,51 +68,36 @@ func (c *BaseClient) NewQuery(q ...any) *Query {
 // It looks a bit weird because of the backward compatibility
 
 func (c *BaseClient) Query(queryContent string, queryVariables interface{}, queryHeaders map[string]interface{}) (any, error) {
-	query := c.NewQuery(queryContent, queryVariables)
 
+	compiledQuery := c.NewQuery(queryContent, queryVariables)
+	parseQueryHeaders, enabledCache, headersModified := c.parseQueryHeaders(queryHeaders)
+
+	var should_cache bool
 	var queryHash string
-	var cachedResponse []byte
-	var cacheBaseClient BaseClient
 
-	// Check for library specific headers
-	if len(queryHeaders) > 0 {
-		cacheBaseClient = reflect.ValueOf(*c).Interface().(BaseClient)
-		queryHeaders = c.parseQueryHeaders(queryHeaders)
-		defer func() {
-			*c = cacheBaseClient
-		}()
+	if headersModified {
+		should_cache = enabledCache
+	} else {
+		should_cache = c.cache.enabled
 	}
 
-	if c.cache.enabled {
-		queryHash = strutil.Md5(fmt.Sprintf("%s-%+v", query.compiledQuery, queryHeaders))
-		c.Logger.Debug(c, "Hash calculated;", "hash:", queryHash)
-
-		cachedResponse = c.cacheLookup(queryHash)
-		if cachedResponse != nil {
-			c.Logger.Debug(c, "Found cached response")
-			return c.decodeResponse(cachedResponse), nil
-		} else {
-			c.Logger.Debug(c, "No cached response found")
-		}
+	if should_cache {
+		queryHash = strutil.Md5(fmt.Sprintf("%s-%+v", compiledQuery.compiledQuery, queryHeaders))
 	}
 
-	response, err := c.executeQuery(query.compiledQuery, queryHeaders)
-	if err != nil {
-		c.Logger.Error(c, "Error while executing query;", "error", err.Error())
-		return nil, err
+	q := &queryExecutor{
+		client:       c,
+		query:        compiledQuery.compiledQuery,
+		headers:      parseQueryHeaders,
+		context:      context.Background(),
+		should_cache: should_cache,
+		hash:         queryHash,
 	}
 
-	jsonData, err := json.Marshal(response)
-	if err != nil {
-		c.Logger.Error(c, "Error while converting to json;", "error", err.Error())
-		return nil, err
-	}
+	q.execute()
+	defer q.done()
 
-	if c.cache.enabled {
-		c.cache.client.Set(queryHash, jsonData)
-	}
-
-	return c.decodeResponse(jsonData), err
+	return q.result.data, q.result.errors
 }
 
 func (c *BaseClient) decodeResponse(jsonData []byte) any {
@@ -137,11 +120,13 @@ func (c *BaseClient) decodeResponse(jsonData []byte) any {
 	}
 }
 
-func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (returnHeaders map[string]interface{}) {
+func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (returnHeaders map[string]interface{}, cache_enabled bool, headers_modified bool) {
 	returnHeaders = make(map[string]interface{})
+
 	for k, v := range queryHeaders {
 		if k == "gqlcache" {
-			c.cache.enabled, _ = strconv.ParseBool(fmt.Sprintf("%v", v))
+			cache_enabled, _ = strconv.ParseBool(fmt.Sprintf("%v", v))
+			headers_modified = true
 			continue
 		}
 		if k == "gqlretries" {
@@ -150,5 +135,5 @@ func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (ret
 		}
 		returnHeaders[k] = v
 	}
-	return returnHeaders
+	return returnHeaders, cache_enabled, headers_modified
 }

query_test.go

@@ -1,6 +1,7 @@
 package gql
 
 import (
+	"math/rand"
 	"reflect"
 	"testing"
 
@@ -297,10 +298,10 @@ func TestBaseClient_parseQueryHeaders(t *testing.T) {
 		queryHeaders map[string]interface{}
 	}
 	tests := []struct {
-		name              string
 		fields            fields
 		args              args
 		wantReturnHeaders map[string]interface{}
+		name              string
 	}{
 		{
 			name:   "Test parseQueryHeaders - no change",
@@ -333,9 +334,199 @@ 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) {
+			if gotReturnHeaders, _, _ := c.parseQueryHeaders(tt.args.queryHeaders); !reflect.DeepEqual(gotReturnHeaders, tt.wantReturnHeaders) {
 				t.Errorf("BaseClient.parseQueryHeaders() = %v, want %v", gotReturnHeaders, tt.wantReturnHeaders)
 			}
 		})
 	}
 }
+
+func TestBaseClient_QueryCache(t *testing.T) {
+	type fields struct {
+		graphql_endpoint string
+		cache_enabled    bool
+	}
+	type args struct {
+		queryVariables interface{}
+		queryHeaders   map[string]interface{}
+		queryContent   string
+	}
+	tests := []struct {
+		want    any
+		args    args
+		name    string
+		fields  fields
+		wantErr bool
+	}{
+		{
+			name: "Test QueryCache - enabled",
+			fields: fields{
+				graphql_endpoint: "https://spacex-production.up.railway.app/",
+				cache_enabled:    true,
+			},
+			args: args{
+				queryContent: `query Dragons {
+					dragons {
+						name
+					}
+				}`,
+				queryHeaders: map[string]interface{}{
+					"x-apollo-operation-name": "Missions-Potato-Test-Golang",
+					"content-type":            "application/json",
+				},
+			},
+		},
+		{
+			name: "Test QueryCache - disabled",
+			fields: fields{
+				graphql_endpoint: "https://spacex-production.up.railway.app/",
+				cache_enabled:    false,
+			},
+			args: args{
+				queryContent: `query Dragons {
+					dragons {
+						name
+					}
+				}`,
+				queryHeaders: map[string]interface{}{
+					"x-apollo-operation-name": "Missions-Potato-Test-Golang",
+					"content-type":            "application/json",
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := NewConnection()
+			c.cache.enabled = tt.fields.cache_enabled
+
+			if tt.fields.graphql_endpoint != "" {
+				c.endpoint = tt.fields.graphql_endpoint
+			}
+
+			_, err := c.Query(tt.args.queryContent, tt.args.queryVariables, tt.args.queryHeaders)
+			if !tt.wantErr {
+				assert.NoError(t, err)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestBaseClient_QueryCacheRandomizedRace(t *testing.T) {
+	type fields struct {
+		graphql_endpoint string
+		cache_enabled    bool
+	}
+	type args struct {
+		queryVariables interface{}
+		queryHeaders   map[string]interface{}
+		queryContent   string
+	}
+	tests := []struct {
+		want    any
+		args    args
+		name    string
+		fields  fields
+		wantErr bool
+	}{
+		{
+			name: "Test QueryCache - enabled",
+			fields: fields{
+				graphql_endpoint: "https://spacex-production.up.railway.app/",
+			},
+			args: args{
+				queryContent: `query Dragons {
+					dragons {
+						name
+					}
+				}`,
+				queryHeaders: map[string]interface{}{
+					"x-apollo-operation-name": "Missions-Potato-Test-Golang",
+					"content-type":            "application/json",
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := NewConnection()
+			if tt.fields.graphql_endpoint != "" {
+				c.endpoint = tt.fields.graphql_endpoint
+			}
+
+			for i := 0; i < 10; i++ {
+				c.cache.enabled = rand.Intn(2) == 0
+				go func() {
+					_, err := c.Query(tt.args.queryContent, tt.args.queryVariables, tt.args.queryHeaders)
+					if !tt.wantErr {
+						assert.NoError(t, err)
+					} else {
+						assert.Error(t, err)
+					}
+				}()
+			}
+		})
+	}
+}
+
+func TestBaseClient_QueryCacheRandomizedRaceViaHeader(t *testing.T) {
+	type fields struct {
+		graphql_endpoint string
+		cache_enabled    bool
+	}
+	type args struct {
+		queryVariables interface{}
+		queryHeaders   map[string]interface{}
+		queryContent   string
+	}
+	tests := []struct {
+		want    any
+		args    args
+		name    string
+		fields  fields
+		wantErr bool
+	}{
+		{
+			name: "Test QueryCache - enabled",
+			fields: fields{
+				graphql_endpoint: "https://spacex-production.up.railway.app/",
+			},
+			args: args{
+				queryContent: `query Dragons {
+					dragons {
+						name
+					}
+				}`,
+				queryHeaders: map[string]interface{}{
+					"x-apollo-operation-name": "Missions-Potato-Test-Golang",
+					"content-type":            "application/json",
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := NewConnection()
+			if tt.fields.graphql_endpoint != "" {
+				c.endpoint = tt.fields.graphql_endpoint
+			}
+
+			for i := 0; i < 10; i++ {
+				tt.args.queryHeaders["gqlcache"] = rand.Intn(2) == 0
+				go func() {
+					_, err := c.Query(tt.args.queryContent, tt.args.queryVariables, tt.args.queryHeaders)
+					if !tt.wantErr {
+						assert.NoError(t, err)
+					} else {
+						assert.Error(t, err)
+					}
+				}()
+			}
+		})
+	}
+}

request.go

@@ -3,6 +3,7 @@ package gql
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"time"
@@ -31,6 +32,7 @@ func (c *BaseClient) executeQuery(query []byte, headers any) (result any, err er
 				c.Logger.Error(c, "Error while executing http request;", "error", err.Error())
 				return err
 			}
+			defer io.Copy(io.Discard, httpResponse.Body) // equivalent to `cp body /dev/null`
 			defer httpResponse.Body.Close()
 
 			if httpResponse.StatusCode <= 200 && httpResponse.StatusCode >= 204 {
@@ -65,3 +67,42 @@ func (c *BaseClient) executeQuery(query []byte, headers any) (result any, err er
 	}
 	return queryResult.Data, nil
 }
+
+func (q *queryExecutor) execute() {
+	if q.should_cache {
+		cachedResponse := q.client.cacheLookup(q.hash)
+		if cachedResponse != nil {
+			q.client.Logger.Debug(q.client, "Found cached response")
+			q.result.data = q.client.decodeResponse(cachedResponse)
+		} else {
+			q.client.Logger.Debug(q.client, "No cached response found")
+		}
+	}
+
+	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())
+		q.result.errors = err
+		return
+	}
+
+	jsonData, err := json.Marshal(response)
+	if err != nil {
+		q.client.Logger.Error(q.client, "Error while converting to json;", "error", err.Error())
+		q.result.errors = err
+		return
+	}
+
+	if q.should_cache {
+		err = q.client.cache.client.Set(q.hash, jsonData)
+		if err != nil {
+			q.client.Logger.Error(q.client, "Error while setting cache key;", "error", err.Error())
+		}
+	}
+
+	q.result.data = q.client.decodeResponse(jsonData)
+}
+
+func (q *queryExecutor) done() {
+	q = nil
+}

structs.go

@@ -44,6 +44,21 @@ type Query struct {
 	compiledQuery []byte         `json:"compiledQuery"`
 }
 
+type queryExecutor struct {
+	result       queryExecutorResult
+	context      context.Context
+	client       *BaseClient
+	headers      map[string]interface{}
+	hash         string
+	query        []byte
+	should_cache bool
+}
+
+type queryExecutorResult struct {
+	data   interface{}
+	errors error
+}
+
 type request struct {
 	Variables any    `json:"variables"`
 	Query     string `json:"query"`