Read, close and replace the http Reponse Body

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

CHANGELOG.md

@@ -30,6 +30,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Removes the need for double error checking ([#246](https://github.com/opensearch-project/opensearch-go/pull/246))
 - Updates workflows to reduce CI time, consolidate OpenSearch versions, update compatibility matrix ([#242](https://github.com/opensearch-project/opensearch-go/pull/242))
 - Moved @svencowart to emeritus maintainers ([#270](https://github.com/opensearch-project/opensearch-go/pull/270))
+- Read, close and replace the http Reponse Body ([#300](https://github.com/opensearch-project/opensearch-go/pull/300))
 
 ### Deprecated
 

opensearchtransport/opensearchtransport.go

@@ -75,13 +75,11 @@ func init() {
 }
 
 // Interface defines the interface for HTTP client.
-//
 type Interface interface {
 	Perform(*http.Request) (*http.Response, error)
 }
 
 // Config represents the configuration of HTTP client.
-//
 type Config struct {
 	URLs     []*url.URL
 	Username string
@@ -113,7 +111,6 @@ type Config struct {
 }
 
 // Client represents the HTTP client.
-//
 type Client struct {
 	sync.Mutex
 
@@ -146,7 +143,6 @@ type Client struct {
 // New creates new transport client.
 //
 // http.DefaultTransport will be used if no transport is passed in the configuration.
-//
 func New(cfg Config) (*Client, error) {
 	if cfg.Transport == nil {
 		cfg.Transport = http.DefaultTransport
@@ -235,7 +231,6 @@ func New(cfg Config) (*Client, error) {
 }
 
 // Perform executes the request and returns a response or error.
-//
 func (c *Client) Perform(req *http.Request) (*http.Response, error) {
 	var (
 		res *http.Response
@@ -407,14 +402,20 @@ func (c *Client) Perform(req *http.Request) (*http.Response, error) {
 			time.Sleep(c.retryBackoff(i + 1))
 		}
 	}
+	// Read, close and replace the http reponse body to close the connection
+	if res != nil && res.Body != nil {
+		body, err := io.ReadAll(res.Body)
+		res.Body.Close()
+		if err == nil {
+			res.Body = io.NopCloser(bytes.NewReader(body))
+		}
+	}
 
 	// TODO(karmi): Wrap error
 	return res, err
 }
 
 // URLs returns a list of transport URLs.
-//
-//
 func (c *Client) URLs() []*url.URL {
 	return c.pool.URLs()
 }

opensearchtransport/opensearchtransport_integration_test.go

@@ -24,6 +24,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build integration
 // +build integration
 
 package opensearchtransport_test
@@ -121,6 +122,33 @@ func TestTransportHeaders(t *testing.T) {
 	}
 }
 
+func TestTransportBodyClose(t *testing.T) {
+	u, _ := url.Parse("http://localhost:9200")
+
+	tp, _ := opensearchtransport.New(opensearchtransport.Config{
+		URLs: []*url.URL{u},
+	})
+
+	req, _ := http.NewRequest("GET", "/", nil)
+	res, err := tp.Perform(req)
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	if closeResp := res.Body.Close(); closeResp != nil {
+		t.Fatalf("Unexpected return on res.Body.Close(): %s", closeResp)
+	}
+	if closeResp := res.Body.Close(); closeResp != nil {
+		t.Fatalf("Unexpected return on res.Body.Close(): %s", closeResp)
+	}
+	body, err := io.ReadAll(res.Body)
+	if err != nil {
+		t.Fatalf("Failed to read the reponse body: %s", err)
+	}
+	if body == nil || len(body) == 0 {
+		t.Fatalf("Unexpected response body:\n%s", body)
+	}
+}
+
 func TestTransportCompression(t *testing.T) {
 	var req *http.Request
 	var res *http.Response