fix(sse): support streaming responses on Fiber/fasthttp adapters (#1059)

* fix(sse): support streaming responses on Fiber/fasthttp adapters

SSE (and other streaming responses) failed on the Fiber adapters with
"unable to flush", because Fiber is built on fasthttp, whose RequestCtx
does not implement http.Flusher and can't flush the response body
synchronously from within the handler; it only streams via
SetBodyStreamWriter.

Add an internal, unexported streaming hook: the sse package detects an
optional bodyStreamer interface on the context (alongside the existing
unwrapper/writeDeadliner) and, when present, streams through the adapter's
callback. The Fiber v2 and v3 adapters implement it via fasthttp's
SetBodyStreamWriter, passing a small writer that satisfies io.Writer,
http.Flusher, and the write-deadline interface so the shared send loop
treats it uniformly. fasthttp stays an indirect dependency and is not
imported by the sse package or the humafiber source.

The SSE send loop is extracted into a shared stream() function reused by
both the direct-write and streaming paths.

Fixes #888.

* test(sse): cover the bodyStreamer streaming path

Add a white-box test that drives Register through a fake huma.Adapter whose
context implements bodyStreamer, exercising the Fiber/fasthttp streaming
branch (which codecov can't credit from the adapter tests, since adapters are
excluded and coverage is measured per package).

Author: wolveix

adapters/humafiber/humafiber.go

@@ -1,11 +1,13 @@
 package humafiber
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"crypto/tls"
 	"io"
 	"mime/multipart"
+	"net"
 	"net/http"
 	"net/url"
 	"strings"
@@ -145,6 +147,16 @@ func (c *fiberWrapper) BodyWriter() io.Writer {
 	return c.orig.RequestCtx()
 }
 
+// StreamBody streams the response body via Fiber/fasthttp's stream writer. It
+// is the optional streaming hook huma's SSE support uses because fasthttp can't
+// flush the response writer synchronously from within the handler.
+func (c *fiberWrapper) StreamBody(fn func(io.Writer)) {
+	rc := c.orig.RequestCtx()
+	rc.SetBodyStreamWriter(func(bw *bufio.Writer) {
+		fn(&fiberStreamWriter{bw: bw, conn: rc.Conn()})
+	})
+}
+
 func (c *fiberWrapper) TLS() *tls.ConnectionState {
 	return c.orig.RequestCtx().TLSConnectionState()
 }
@@ -262,3 +274,26 @@ func New(r *fiber.App, config huma.Config) huma.API {
 func NewWithGroup(r *fiber.App, g fiber.Router, config huma.Config) huma.API {
 	return huma.NewAPI(config, &fiberAdapter{tester: r, router: g})
 }
+
+// fiberStreamWriter adapts fasthttp's buffered stream writer to the io.Writer,
+// http.Flusher, and write-deadline interfaces the streaming code expects. It is
+// shared by the Fiber v2 and v3 adapters.
+type fiberStreamWriter struct {
+	bw   *bufio.Writer
+	conn net.Conn
+}
+
+func (w *fiberStreamWriter) Write(p []byte) (int, error) {
+	return w.bw.Write(p)
+}
+
+func (w *fiberStreamWriter) Flush() {
+	_ = w.bw.Flush()
+}
+
+func (w *fiberStreamWriter) SetWriteDeadline(t time.Time) error {
+	if w.conn == nil {
+		return nil
+	}
+	return w.conn.SetWriteDeadline(t)
+}

adapters/humafiber/humafiber_test.go

@@ -2,14 +2,60 @@ package humafiber
 
 import (
 	"context"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
 	"github.com/danielgtaylor/huma/v2"
+	"github.com/danielgtaylor/huma/v2/sse"
 	"github.com/gofiber/fiber/v3"
 )
 
+// TestSSE exercises the Fiber-specific streaming path (the StreamBody hook /
+// fasthttp SetBodyStreamWriter) so Server-Sent Events flush to the client
+// instead of failing with "unable to flush". Regression test for #888.
+func TestSSE(t *testing.T) {
+	app := fiber.New()
+	api := New(app, huma.DefaultConfig("Test", "1.0.0"))
+
+	type Message struct {
+		Text string `json:"text"`
+	}
+
+	sse.Register(api, huma.Operation{
+		OperationID: "sse",
+		Method:      http.MethodGet,
+		Path:        "/sse",
+	}, map[string]any{
+		"message": Message{},
+	}, func(ctx context.Context, input *struct{}, send sse.Sender) {
+		_ = send.Data(Message{Text: "hello"})
+		_ = send.Comment("heartbeat")
+	})
+
+	resp, err := app.Test(httptest.NewRequest(http.MethodGet, "http://example.com/sse", nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200", resp.StatusCode)
+	}
+	if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" {
+		t.Fatalf("Content-Type = %q, want text/event-stream", ct)
+	}
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := "data: {\"text\":\"hello\"}\n\n: heartbeat\n\n"
+	if string(body) != want {
+		t.Fatalf("body = %q, want %q", body, want)
+	}
+}
+
 // TestNewContext covers the exported NewContext constructor. Unlike the other
 // adapters, the Fiber adapter's Handle wraps the context to expose fasthttp
 // user values, so it can't call NewContext itself; exercise it directly here.

adapters/humafiber/humafiber_v2.go

@@ -1,6 +1,7 @@
 package humafiber
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"crypto/tls"
@@ -144,6 +145,16 @@ func (c *fiberV2Wrapper) BodyWriter() io.Writer {
 	return c.orig.Context()
 }
 
+// StreamBody streams the response body via Fiber/fasthttp's stream writer. It
+// is the optional streaming hook huma's SSE support uses because fasthttp can't
+// flush the response writer synchronously from within the handler.
+func (c *fiberV2Wrapper) StreamBody(fn func(io.Writer)) {
+	rc := c.orig.Context()
+	rc.SetBodyStreamWriter(func(bw *bufio.Writer) {
+		fn(&fiberStreamWriter{bw: bw, conn: rc.Conn()})
+	})
+}
+
 func (c *fiberV2Wrapper) TLS() *tls.ConnectionState {
 	return c.orig.Context().TLSConnectionState()
 }

adapters/humafiber/humafiber_v2_test.go

@@ -9,10 +9,45 @@ import (
 	"testing"
 
 	"github.com/danielgtaylor/huma/v2"
+	"github.com/danielgtaylor/huma/v2/sse"
 	fiberV2 "github.com/gofiber/fiber/v2"
 	"github.com/stretchr/testify/require"
 )
 
+// TestSSEV2 exercises the Fiber v2 streaming path (the StreamBody hook /
+// fasthttp SetBodyStreamWriter) so Server-Sent Events flush to the client.
+// Regression test for #888.
+func TestSSEV2(t *testing.T) {
+	app := fiberV2.New()
+	api := NewV2(app, huma.DefaultConfig("Test", "1.0.0"))
+
+	type Message struct {
+		Text string `json:"text"`
+	}
+
+	sse.Register(api, huma.Operation{
+		OperationID: "sse",
+		Method:      http.MethodGet,
+		Path:        "/sse",
+	}, map[string]any{
+		"message": Message{},
+	}, func(ctx context.Context, input *struct{}, send sse.Sender) {
+		_ = send.Data(Message{Text: "hello"})
+		_ = send.Comment("heartbeat")
+	})
+
+	req, _ := http.NewRequest(http.MethodGet, "http://example.com/sse", nil)
+	resp, err := app.Test(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.Equal(t, "data: {\"text\":\"hello\"}\n\n: heartbeat\n\n", string(body))
+}
+
 // TestFiberV2EachHeaderAndCookie ensures EachHeader yields one callback per
 // header value (not one per byte) so header-based helpers like huma.ReadCookie
 // continue to work under the Fiber v2 adapter. Regression test for #1055.