Improvement: reuse RPC connection

dipdup-io · Aug 18, 2023 · b2736a5 · b2736a5
1 parent fb7a6f9
commit b2736a5
Show file tree

Hide file tree

Showing 4 changed files with 34 additions and 15 deletions.
diff --git a/pkg/rpc/api.go b/pkg/rpc/api.go
@@ -20,9 +20,9 @@ type API struct {
 // NewAPI - constructor of API
 func NewAPI(baseURL string, opts ...ApiOption) API {
 	t := http.DefaultTransport.(*http.Transport).Clone()
-	t.MaxIdleConns = 100
-	t.MaxConnsPerHost = 100
-	t.MaxIdleConnsPerHost = 100
+	t.MaxIdleConns = 10
+	t.MaxConnsPerHost = 10
+	t.MaxIdleConnsPerHost = 10
 
 	client := &http.Client{
 		Transport: t,

diff --git a/pkg/rpc/post.go b/pkg/rpc/post.go
@@ -3,6 +3,7 @@ package api
 import (
 	"bytes"
 	"context"
+	"io"
 	"net/http"
 
 	"github.com/goccy/go-json"
@@ -33,11 +34,16 @@ func post[T any](ctx context.Context, api API, req Request, output *Response[T])
 	}
 	defer response.Body.Close()
 
+	buffer := new(bytes.Buffer)
+	if _, err := io.Copy(buffer, response.Body); err != nil {
+		return err
+	}
+
 	if response.StatusCode != http.StatusOK {
 		return errors.Wrapf(ErrRequest, "request %d invalid status code: %d", output.ID, response.StatusCode)
 	}
 
-	if err := json.NewDecoder(response.Body).Decode(output); err != nil {
+	if err := json.NewDecoder(buffer).Decode(output); err != nil {
 		return err
 	}
 

diff --git a/pkg/sequencer/api.go b/pkg/sequencer/api.go
@@ -21,20 +21,12 @@ type API struct {
 	feederGatewayUrl string
 	cacheDir         string
 	rateLimit        *rate.Limiter
+	rps              int
 }
 
 // NewAPI - constructor of API
 func NewAPI(gatewayUrl, feederGatewayUrl string, opts ...ApiOption) API {
-	t := http.DefaultTransport.(*http.Transport).Clone()
-	t.MaxIdleConns = 100
-	t.MaxConnsPerHost = 100
-	t.MaxIdleConnsPerHost = 100
-
-	client := &http.Client{
-		Transport: t,
-	}
 	api := API{
-		client:           client,
 		gatewayUrl:       gatewayUrl,
 		feederGatewayUrl: feederGatewayUrl,
 	}
@@ -43,6 +35,21 @@ func NewAPI(gatewayUrl, feederGatewayUrl string, opts ...ApiOption) API {
 		opts[i](&api)
 	}
 
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	if api.rps < 1 || api.rps > 100 {
+		t.MaxIdleConns = 10
+		t.MaxConnsPerHost = 10
+		t.MaxIdleConnsPerHost = 10
+	} else {
+		t.MaxIdleConns = api.rps
+		t.MaxConnsPerHost = api.rps
+		t.MaxIdleConnsPerHost = api.rps
+	}
+
+	api.client = &http.Client{
+		Transport: t,
+	}
+
 	return api
 }
 
@@ -153,14 +160,19 @@ func (api API) post(ctx context.Context, baseURL, path string, args map[string]s
 	}
 	defer response.Body.Close()
 
+	buffer := new(bytes.Buffer)
+	if _, err := io.Copy(buffer, response.Body); err != nil {
+		return err
+	}
+
 	if response.StatusCode != http.StatusOK {
 		var e Error
-		if err := json.NewDecoder(response.Body).Decode(&e); err != nil {
+		if err := json.NewDecoder(buffer).Decode(&e); err != nil {
 			return errors.Wrap(ErrRequest, err.Error())
 		}
 		return errors.Wrap(ErrRequest, e.Error())
 	}
 
-	err = json.NewDecoder(response.Body).Decode(output)
+	err = json.NewDecoder(buffer).Decode(output)
 	return err
 }
diff --git a/pkg/sequencer/options.go b/pkg/sequencer/options.go
@@ -14,6 +14,7 @@ func WithRateLimit(requestPerSecond int) ApiOption {
 	return func(api *API) {
 		if requestPerSecond > 0 {
 			api.rateLimit = rate.NewLimiter(rate.Every(time.Second/time.Duration(requestPerSecond)), requestPerSecond)
+			api.rps = requestPerSecond
 		}
 	}
 }