Add stream response body support

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.Greater(t, n, 0)
+
+	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,10 +81,16 @@ func (c *core) execFunc() (*Response, error) {
 	c.req.RawRequest.CopyTo(reqv)
 	cfg := c.getRetryConfig()
 
+	// Set StreamResponseBody option from request setting (falls back to client setting)
+	originalStreamResponseBody := c.client.fasthttp.StreamResponseBody
+	c.client.fasthttp.StreamResponseBody = c.req.StreamResponseBody()
+
 	var err error
 	go func() {
 		respv := fasthttp.AcquireResponse()
 		defer func() {
+			// Restore original StreamResponseBody setting
+			c.client.fasthttp.StreamResponseBody = originalStreamResponseBody
 			fasthttp.ReleaseRequest(reqv)
 			fasthttp.ReleaseResponse(respv)
 		}()

client/request.go

@@ -67,7 +67,8 @@ type Request struct {
 	timeout      time.Duration
 	maxRedirects int
 
-	bodyType bodyType
+	bodyType           bodyType
+	streamResponseBody *bool // nil means use client setting
 }
 
 // Method returns the HTTP method set in the Request.
@@ -590,6 +591,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 +676,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,17 @@ 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())
+}
+
 // String returns the response body as a trimmed string.
 func (r *Response) String() string {
 	return utils.Trim(string(r.Body()), ' ')

client/response_test.go

@@ -538,3 +538,75 @@ func Test_Response_Save(t *testing.T) {
 		require.Error(t, err)
 	})
 }
+
+func Test_Response_BodyStream(t *testing.T) {
+	t.Parallel()
+
+	server := startTestServer(t, func(app *fiber.App) {
+		app.Get("/stream", func(c fiber.Ctx) error {
+			return c.SendString("streaming data")
+		})
+		app.Get("/large", func(c fiber.Ctx) error {
+			data := make([]byte, 1024)
+			for i := range data {
+				data[i] = byte('A' + i%26)
+			}
+			return c.Send(data)
+		})
+	})
+	defer server.stop()
+
+	t.Run("basic streaming", func(t *testing.T) {
+		client := New().SetDial(server.dial()).SetStreamResponseBody(true)
+
+		resp, err := client.Get("http://example.com/stream")
+		require.NoError(t, err)
+		defer resp.Close()
+		bodyStream := resp.BodyStream()
+		require.NotNil(t, bodyStream)
+		data, err := io.ReadAll(bodyStream)
+		require.NoError(t, err)
+		require.Equal(t, "streaming data", string(data))
+	})
+
+	t.Run("large response streaming", func(t *testing.T) {
+		client := New().SetDial(server.dial()).SetStreamResponseBody(true)
+		resp, err := client.Get("http://example.com/large")
+		require.NoError(t, err)
+		defer resp.Close()
+		bodyStream := resp.BodyStream()
+		require.NotNil(t, bodyStream)
+		buffer := make([]byte, 256)
+		var totalRead []byte
+		for {
+			n, err := bodyStream.Read(buffer)
+			if n > 0 {
+				totalRead = append(totalRead, buffer[:n]...)
+			}
+			if err == io.EOF {
+				break
+			}
+			require.NoError(t, err)
+		}
+		require.Equal(t, 1024, len(totalRead))
+		for i := 0; i < 1024; i++ {
+			expected := byte('A' + i%26)
+			require.Equal(t, expected, totalRead[i])
+		}
+	})
+
+	t.Run("compare with regular body", func(t *testing.T) {
+		client1 := New().SetDial(server.dial())
+		resp1, err := client1.Get("http://example.com/stream")
+		require.NoError(t, err)
+		defer resp1.Close()
+		normalBody := resp1.Body()
+		client2 := New().SetDial(server.dial()).SetStreamResponseBody(true)
+		resp2, err := client2.Get("http://example.com/stream")
+		require.NoError(t, err)
+		defer resp2.Close()
+		streamedBody, err := io.ReadAll(resp2.BodyStream())
+		require.NoError(t, err)
+		require.Equal(t, normalBody, streamedBody)
+	})
+}