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

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.