Skip to content
Docs
Guides
Server Streaming

Server Streaming

Sphere maps protobuf server-streaming methods to Server-Sent Events (SSE). The generated HTTP endpoint works through httpx on Gin, Echo, Hertz, and Fiber; it is independent of any gRPC stream generated by protoc-gen-go-grpc.

Define a Server Stream

Declare stream on the response and keep the usual google.api.http request mapping:

syntax = "proto3";

package chat.v1;

import "google/api/annotations.proto";

service ChatService {
  rpc Chat(ChatRequest) returns (stream ChatResponse) {
    option (google.api.http) = {
      post: "/api/chat"
      body: "*"
    };
  }
}

message ChatRequest {
  string prompt = 1;
}

message ChatResponse {
  string delta = 1;
}

protoc-gen-sphere generates this service contract:

type ChatServiceHTTPServer interface {
    Chat(context.Context, *ChatRequest, func(*ChatResponse) error) error
}

Only server-streaming is mapped to SSE. Client-streaming and bidirectional RPCs are skipped with a warning, or fail generation when fail_on_warn is enabled.

Implement the Producer

Call send once for each reply. Stop promptly when send fails or the request context is canceled; both normally mean the client can no longer receive data.

func (s *Service) Chat(
    ctx context.Context,
    req *chatv1.ChatRequest,
    send func(*chatv1.ChatResponse) error,
) error {
    for _, delta := range tokenize(req.Prompt) {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }
        if err := send(&chatv1.ChatResponse{Delta: delta}); err != nil {
            return err
        }
    }
    return nil
}

The producer runs independently from the HTTP adapter callback. Do not retain or use httpx.Context in it; use the standard context.Context, the bound request, and ordinary service dependencies.

Wire Contract

With the default httpz.WithSSE wrapper, reply messages are unnamed SSE events whose data field uses the same JSON encoding as unary responses. A successful stream ends with a done event:

data: {"delta":"hello"}

data: {"delta":" world"}

event: done
data: {}

A failed stream ends in one of two ways:

  • Binding, validation, preparation, or producer failure before the first reply: a regular JSON error response with its normal HTTP status.
  • Failure after the first reply: HTTP 200 is already committed, so the stream ends with an error event carrying the standard ErrorResponse JSON.

If neither done nor error arrives, treat the stream as interrupted. After the response commits, the default wrapper sends comment frames every 15 seconds to keep an idle connection alive and sets Cache-Control: no-cache plus X-Accel-Buffering: no.

Idle Streams and Eager Commit

The generated default is lazy: it waits for the first reply before committing the SSE response. This preserves the option to return an ordinary HTTP error if the producer fails before sending, but an endpoint that waits indefinitely for its first live event has not started its heartbeat yet.

For that push-style endpoint, configure stream_handler_func with a project wrapper that enables eager commit:

func WithEagerSSE[T any](
    prepare func(httpx.Context) (httpz.SSEStream[T], error),
) httpx.Handler {
    return httpz.WithSSE(prepare, httpz.WithSSEEagerCommit())
}

Eager mode commits HTTP 200 immediately, so heartbeats can run before the first reply. In exchange, every producer failure is an in-stream error event rather than a non-200 response.

Request and Resume Design

Request binding follows the same rules as unary methods. GET streams typically bind filters from URI/query/header fields; POST streams may bind a JSON body. response_body is ignored with a warning because each event contains a whole reply message.

The default httpz.WithSSE wrapper does not emit SSE id: fields. If an endpoint supports resume, make replay/cursor semantics explicit in the reply, bind Last-Event-ID with BINDING_LOCATION_HEADER (or use a query cursor), and provide a custom stream_handler_func when transport-level SSE IDs are required.

Browser EventSource only opens GET requests. For POST streams, use streaming fetch or another client capable of parsing SSE frames.

Operations and Testing

  • Exclude SSE routes from response-buffering middleware such as gzip.
  • Configure server and reverse-proxy timeouts for the intended stream lifetime.
  • Test incremental delivery and disconnect behavior over a real HTTP connection; httpx.TestRequester intentionally buffers the final stream body.
  • Gin, Echo, and Hertz also expose httpx.Flusher; Fiber does not. Prefer httpx.Streamer or httpz.WithSSE for portable streaming.

Related

Last updated on
LoggingHTTP Runtime