🔥 feat: Add StreamResponseBody support for the Client

Author: Abhirup-99

client/client.go

@@ -55,9 +55,10 @@ type Client struct {
 	userResponseHooks    []ResponseHook
 	builtinResponseHooks []ResponseHook
 
-	timeout time.Duration
-	mu      sync.RWMutex
-	debug   bool
+	timeout            time.Duration
+	mu                 sync.RWMutex
+	debug              bool
+	streamResponseBody bool
 }
 
 // R creates a new Request associated with the client.
@@ -435,6 +436,20 @@ func (c *Client) DisableDebug() *Client {
 	return c
 }
 
+// StreamResponseBody returns the current StreamResponseBody setting.
+func (c *Client) StreamResponseBody() bool {
+	return c.streamResponseBody
+}
+
+// SetStreamResponseBody enables or disables response body streaming.
+// When enabled, the response body can be read as a stream using BodyStream()
+// instead of being fully loaded into memory. This is useful for large responses
+// or server-sent events.
+func (c *Client) SetStreamResponseBody(enable bool) *Client {
+	c.streamResponseBody = enable
+	return c
+}
+
 // SetCookieJar sets the cookie jar for the client.
 func (c *Client) SetCookieJar(cookieJar *CookieJar) *Client {
 	c.cookieJar = cookieJar

client/client_test.go

@@ -1733,3 +1733,151 @@ func Benchmark_Client_Request_Parallel(b *testing.B) {
 		require.NoError(b, err)
 	})
 }
+
+func Test_Client_StreamResponseBody(t *testing.T) {
+	t.Parallel()
+	client := New()
+	require.False(t, client.StreamResponseBody())
+	client.SetStreamResponseBody(true)
+	require.True(t, client.StreamResponseBody())
+	client.SetStreamResponseBody(false)
+	require.False(t, client.StreamResponseBody())
+}
+
+func Test_Client_StreamResponseBody_ServerSentEvents(t *testing.T) {
+	t.Parallel()
+
+	app, addr := startTestServerWithPort(t, func(app *fiber.App) {
+		app.Get("/sse", func(c fiber.Ctx) error {
+			c.Set("Content-Type", "text/event-stream")
+			c.Set("Cache-Control", "no-cache")
+			c.Set("Connection", "keep-alive")
+
+			messages := []string{
+				"data: message 1\n\n",
+				"data: message 2\n\n",
+				"data: message 3\n\n",
+			}
+
+			for _, msg := range messages {
+				if _, err := c.WriteString(msg); err != nil {
+					return err
+				}
+			}
+
+			return nil
+		})
+	})
+	defer func() { require.NoError(t, app.Shutdown()) }()
+
+	client := New().SetStreamResponseBody(true)
+	resp, err := client.Get("http://" + addr + "/sse")
+	require.NoError(t, err)
+	defer resp.Close()
+
+	bodyStream := resp.BodyStream()
+	require.NotNil(t, bodyStream)
+
+	buffer := make([]byte, 1024)
+	n, err := bodyStream.Read(buffer)
+	require.NoError(t, err)
+	require.Positive(t, n)
+
+	content := string(buffer[:n])
+	require.Contains(t, content, "data: message 1")
+}
+
+func Test_Client_StreamResponseBody_LargeResponse(t *testing.T) {
+	t.Parallel()
+
+	largeData := make([]byte, 1024*1024)
+	for i := range largeData {
+		largeData[i] = byte(i % 256)
+	}
+
+	app, addr := startTestServerWithPort(t, func(app *fiber.App) {
+		app.Get("/large", func(c fiber.Ctx) error {
+			return c.Send(largeData)
+		})
+	})
+	defer func() { require.NoError(t, app.Shutdown()) }()
+	client := New().SetStreamResponseBody(true)
+	resp, err := client.Get("http://" + addr + "/large")
+	require.NoError(t, err)
+	defer resp.Close()
+	bodyStream := resp.BodyStream()
+	require.NotNil(t, bodyStream)
+	streamedData, err := io.ReadAll(bodyStream)
+	require.NoError(t, err)
+	require.Equal(t, largeData, streamedData)
+	client2 := New()
+	resp2, err := client2.Get("http://" + addr + "/large")
+	require.NoError(t, err)
+	defer resp2.Close()
+	body := resp2.Body()
+	require.Equal(t, largeData, body)
+}
+
+func Test_Client_StreamResponseBody_Disabled_Default(t *testing.T) {
+	t.Parallel()
+
+	app, addr := startTestServerWithPort(t, func(app *fiber.App) {
+		app.Get("/test", func(c fiber.Ctx) error {
+			return c.SendString("Hello, World!")
+		})
+	})
+	defer func() { require.NoError(t, app.Shutdown()) }()
+
+	client := New()
+	resp, err := client.Get("http://" + addr + "/test")
+	require.NoError(t, err)
+	defer resp.Close()
+
+	body := resp.Body()
+	require.Equal(t, "Hello, World!", string(body))
+
+	bodyStream := resp.BodyStream()
+	require.NotNil(t, bodyStream)
+}
+
+func Test_Client_StreamResponseBody_ChainableMethods(t *testing.T) {
+	t.Parallel()
+
+	client := New().
+		SetStreamResponseBody(true).
+		SetTimeout(time.Second * 5).
+		SetStreamResponseBody(false)
+
+	require.False(t, client.StreamResponseBody())
+}
+
+func Test_Request_StreamResponseBody(t *testing.T) {
+	t.Parallel()
+
+	app, addr := startTestServerWithPort(t, func(app *fiber.App) {
+		app.Get("/test", func(c fiber.Ctx) error {
+			return c.SendString("Hello, World!")
+		})
+	})
+	defer func() { require.NoError(t, app.Shutdown()) }()
+
+	client := New().SetStreamResponseBody(false) // client has streaming disabled
+	req := client.R().SetStreamResponseBody(true)
+	require.True(t, req.StreamResponseBody())
+
+	resp, err := req.Get("http://" + addr + "/test")
+	require.NoError(t, err)
+	defer resp.Close()
+	bodyStream := resp.BodyStream()
+	require.NotNil(t, bodyStream)
+	req2 := client.R().SetStreamResponseBody(false)
+	require.False(t, req2.StreamResponseBody())
+	clientWithStreaming := New().SetStreamResponseBody(true)
+	req3 := clientWithStreaming.R()
+	require.True(t, req3.StreamResponseBody()) // Should inherit from client
+	req4 := client.R().
+		SetStreamResponseBody(true).
+		SetTimeout(time.Second * 5).
+		SetStreamResponseBody(false)
+	require.False(t, req4.StreamResponseBody())
+}

client/core.go

@@ -81,6 +81,43 @@ func (c *core) execFunc() (*Response, error) {
 	c.req.RawRequest.CopyTo(reqv)
 	cfg := c.getRetryConfig()
 
+	// Determine which client to use - create a new one if StreamResponseBody differs
+	var fastHTTPClient *fasthttp.Client
+	requestStreamResponseBody := c.req.StreamResponseBody()
+	c.client.mu.RLock()
+	clientStream := c.client.streamResponseBody
+	original := c.client.fasthttp
+
+	if requestStreamResponseBody != clientStream {
+		// Request setting differs from client setting, create a temporary client
+
+		fastHTTPClient = &fasthttp.Client{
+			Dial:                          original.Dial,
+			DialDualStack:                 original.DialDualStack,
+			TLSConfig:                     original.TLSConfig,
+			MaxConnsPerHost:               original.MaxConnsPerHost,
+			MaxIdleConnDuration:           original.MaxIdleConnDuration,
+			MaxConnDuration:               original.MaxConnDuration,
+			ReadTimeout:                   original.ReadTimeout,
+			WriteTimeout:                  original.WriteTimeout,
+			ReadBufferSize:                original.ReadBufferSize,
+			WriteBufferSize:               original.WriteBufferSize,
+			MaxResponseBodySize:           original.MaxResponseBodySize,
+			NoDefaultUserAgentHeader:      original.NoDefaultUserAgentHeader,
+			DisableHeaderNamesNormalizing: original.DisableHeaderNamesNormalizing,
+			DisablePathNormalizing:        original.DisablePathNormalizing,
+			MaxIdemponentCallAttempts:     original.MaxIdemponentCallAttempts,
+			Name:                          original.Name,
+			ConfigureClient:               original.ConfigureClient,
+
+			// Request-specific override
+			StreamResponseBody: requestStreamResponseBody,
+		}
+	} else {
+		fastHTTPClient = original
+	}
+	c.client.mu.RUnlock()
+
 	var err error
 	go func() {
 		respv := fasthttp.AcquireResponse()
@@ -93,15 +130,15 @@ func (c *core) execFunc() (*Response, error) {
 			// Use an exponential backoff retry strategy.
 			err = retry.NewExponentialBackoff(*cfg).Retry(func() error {
 				if c.req.maxRedirects > 0 && (string(reqv.Header.Method()) == fiber.MethodGet || string(reqv.Header.Method()) == fiber.MethodHead) {
-					return c.client.fasthttp.DoRedirects(reqv, respv, c.req.maxRedirects)
+					return fastHTTPClient.DoRedirects(reqv, respv, c.req.maxRedirects)
 				}
-				return c.client.fasthttp.Do(reqv, respv)
+				return fastHTTPClient.Do(reqv, respv)
 			})
 		} else {
 			if c.req.maxRedirects > 0 && (string(reqv.Header.Method()) == fiber.MethodGet || string(reqv.Header.Method()) == fiber.MethodHead) {
-				err = c.client.fasthttp.DoRedirects(reqv, respv, c.req.maxRedirects)
+				err = fastHTTPClient.DoRedirects(reqv, respv, c.req.maxRedirects)
 			} else {
-				err = c.client.fasthttp.Do(reqv, respv)
+				err = fastHTTPClient.Do(reqv, respv)
 			}
 		}
 

client/request.go

@@ -44,30 +44,25 @@ var ErrClientNil = errors.New("client cannot be nil")
 
 // Request contains all data related to an HTTP request.
 type Request struct {
-	ctx context.Context //nolint:containedctx // Context is needed to be stored in the request.
-
-	body    any
-	header  *Header
-	params  *QueryParam
-	cookies *Cookie
-	path    *PathParam
-
-	client *Client
-
-	formData *FormData
-
-	RawRequest *fasthttp.Request
-	url        string
-	method     string
-	userAgent  string
-	boundary   string
-	referer    string
-	files      []*File
-
-	timeout      time.Duration
-	maxRedirects int
-
-	bodyType bodyType
+	files              []*File
+	ctx                context.Context //nolint:containedctx // Context is needed to be stored in the request.
+	body               any
+	url                string
+	method             string
+	userAgent          string
+	boundary           string
+	referer            string
+	header             *Header
+	params             *QueryParam
+	cookies            *Cookie
+	path               *PathParam
+	client             *Client
+	formData           *FormData
+	RawRequest         *fasthttp.Request
+	streamResponseBody *bool // nil means use client setting
+	timeout            time.Duration
+	maxRedirects       int
+	bodyType           bodyType
 }
 
 // Method returns the HTTP method set in the Request.
@@ -590,6 +585,25 @@ func (r *Request) SetMaxRedirects(count int) *Request {
 	return r
 }
 
+// StreamResponseBody returns the StreamResponseBody setting for this request.
+// Returns the client's setting if not explicitly set at the request level.
+func (r *Request) StreamResponseBody() bool {
+	if r.streamResponseBody != nil {
+		return *r.streamResponseBody
+	}
+	if r.client != nil {
+		return r.client.streamResponseBody
+	}
+	return false
+}
+
+// SetStreamResponseBody sets the StreamResponseBody option for this specific request,
+// overriding the client-level setting.
+func (r *Request) SetStreamResponseBody(enable bool) *Request {
+	r.streamResponseBody = &enable
+	return r
+}
+
 // checkClient ensures that a Client is set. If none is set, it defaults to the global defaultClient.
 func (r *Request) checkClient() {
 	if r.client == nil {
@@ -656,6 +670,7 @@ func (r *Request) Reset() {
 	r.maxRedirects = 0
 	r.bodyType = noBody
 	r.boundary = boundary
+	r.streamResponseBody = nil
 
 	for len(r.files) != 0 {
 		t := r.files[0]

client/response.go

@@ -89,6 +89,22 @@ func (r *Response) Body() []byte {
 	return r.RawResponse.Body()
 }
 
+// BodyStream returns the response body as a stream reader.
+// Note: When using BodyStream(), the response body is not copied to memory,
+// so calling Body() afterwards may return an empty slice.
+func (r *Response) BodyStream() io.Reader {
+	if stream := r.RawResponse.BodyStream(); stream != nil {
+		return stream
+	}
+	// If streaming is not enabled, return a bytes.Reader from the regular body
+	return bytes.NewReader(r.RawResponse.Body())
+}
+
+// IsStreaming returns true if the response body is being streamed.
+func (r *Response) IsStreaming() bool {
+	return r.RawResponse.BodyStream() != nil
+}
+
 // String returns the response body as a trimmed string.
 func (r *Response) String() string {
 	return utils.Trim(string(r.Body()), ' ')
@@ -143,14 +159,17 @@ func (r *Response) Save(v any) error {
 		return nil
 
 	case io.Writer:
-		if _, err := io.Copy(p, bytes.NewReader(r.Body())); err != nil {
-			return fmt.Errorf("failed to write response body to io.Writer: %w", err)
+		var err error
+		if r.IsStreaming() {
+			_, err = io.Copy(p, r.BodyStream())
+		} else {
+			_, err = io.Copy(p, bytes.NewReader(r.Body()))
 		}
-		defer func() {
-			if pc, ok := p.(io.WriteCloser); ok {
-				_ = pc.Close() //nolint:errcheck // not needed
-			}
-		}()
+
+		if err != nil {
+			return fmt.Errorf("failed to write response body to writer: %w", err)
+		}
+
 		return nil
 
 	default: