Go Streaming Patterns for SSE #Permalink to this section

Part of Backend Stream Generation & Connection Management.

Go’s concurrency primitives — goroutines, channels, and context.Context — map directly onto the demands of SSE: a persistent HTTP connection that flushes data as events arrive, tears down cleanly when the client disconnects, and scales to tens of thousands of concurrent streams without per-connection thread overhead. The core challenge is wiring Go’s http.Flusher interface, unbuffered or buffered channels, and context cancellation into a pattern that is both correct under race conditions and observable in production.

This guide covers the full arc: the wire protocol and flush mechanics, a reference http.HandlerFunc implementation, fan-out via a hub, graceful shutdown, edge cases around proxies and CDNs, and the performance knobs that matter at scale.

How It Works #Permalink to this section

Go’s net/http server dispatches each incoming request to a handler running in its own goroutine. An SSE handler must keep that goroutine alive for the duration of the connection, writing text/event-stream frames and calling Flush() after each one.

The http.Flusher interface #Permalink to this section

http.ResponseWriter alone buffers output. http.Flusher exposes the single method Flush() that sends whatever is currently in the response buffer down the wire. You type-assert to it at handler startup:

flusher, ok := w.(http.Flusher)
if !ok {
    http.Error(w, "streaming not supported", http.StatusInternalServerError)
    return
}

Failure means the response writer is wrapped (e.g., by some middleware) without forwarding Flush. Common culprit: gzip middleware. Either bypass it for SSE routes or use a flusher-aware gzip wrapper.

Wire format recap #Permalink to this section

Each SSE event is one or more field lines followed by a blank line:

data: {"temp": 21.3, "unit": "C"}\n
id: 1718966400001\n
event: sensor\n
\n

The \n after each field and the blank line terminator are mandatory per the WHATWG SSE spec. Go’s fmt.Fprintf into a ResponseWriter is byte-exact, so use explicit \n — not \r\n. The text/event-stream MIME type requires Unix line endings.

For multi-line payloads, prefix each line with data: . See Formatting Multi-Line data Fields in SSE for the full encoding rules.

The id field drives client reconnection; the Event ID & Retry Mechanism Design guide covers how Last-Event-ID is replayed on reconnect.

Server-Side Implementation #Permalink to this section

Minimal single-handler pattern #Permalink to this section

package main

import (
    "fmt"
    "net/http"
    "time"
)

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // 1. Assert flusher support.
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "streaming unsupported", http.StatusInternalServerError)
        return
    }

    // 2. Required headers — must be set before the first Write.
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    // Disable nginx / proxy buffering (see Edge Cases below).
    w.Header().Set("X-Accel-Buffering", "no")

    // 3. Send an initial retry hint (milliseconds).
    fmt.Fprintf(w, "retry: 3000\n\n")
    flusher.Flush()

    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    var seq int64
    for {
        select {
        case <-r.Context().Done(): // client disconnected or server shutdown
            return
        case t := <-ticker.C:
            seq++
            fmt.Fprintf(w, "id: %d\n", seq)
            fmt.Fprintf(w, "event: tick\n")
            fmt.Fprintf(w, "data: {\"ts\":%d}\n\n", t.UnixMilli())
            flusher.Flush()
        }
    }
}

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/events", sseHandler)
    srv := &http.Server{Addr: ":8080", Handler: mux}
    if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
        panic(err)
    }
}

r.Context().Done() is closed automatically when the TCP connection drops, giving you a zero-cost disconnect signal without polling w.Write for errors.

Hub fan-out pattern #Permalink to this section

For broadcast scenarios — price tickers, chat rooms, live dashboards — a central hub serialises registrations and distributes events through per-client channels. This avoids holding a mutex during slow Write/Flush calls.

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log/slog"
    "net/http"
    "sync"
    "time"
)

// Event is the payload sent to each subscriber.
type Event struct {
    ID   int64           `json:"id"`
    Type string          `json:"type"`
    Data json.RawMessage `json:"data"`
}

// client represents one SSE connection.
type client struct {
    send chan Event
    done <-chan struct{} // mirrors r.Context().Done()
}

// Hub serialises registration and broadcast without blocking on I/O.
type Hub struct {
    mu      sync.Mutex
    clients map[*client]struct{}
}

func NewHub() *Hub { return &Hub{clients: make(map[*client]struct{})} }

func (h *Hub) register(c *client) {
    h.mu.Lock()
    h.clients[c] = struct{}{}
    h.mu.Unlock()
}

func (h *Hub) unregister(c *client) {
    h.mu.Lock()
    delete(h.clients, c)
    h.mu.Unlock()
    close(c.send) // signal the handler's write loop to stop
}

// Broadcast fans out to every connected client.
// Non-blocking send: slow clients are dropped, not blocked on.
func (h *Hub) Broadcast(e Event) {
    h.mu.Lock()
    for c := range h.clients {
        select {
        case c.send <- e:
        default:
            // client channel full — drop event for this subscriber
            slog.Warn("broadcast: dropped event", "client_ptr", fmt.Sprintf("%p", c))
        }
    }
    h.mu.Unlock()
}

// SSEHandler registers the caller as a subscriber and streams events.
func (h *Hub) SSEHandler(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "streaming unsupported", http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("X-Accel-Buffering", "no")

    c := &client{
        send: make(chan Event, 64), // buffer 64 events per client
        done: r.Context().Done(),
    }
    h.register(c)
    defer h.unregister(c)

    // Flush initial retry directive.
    fmt.Fprintf(w, "retry: 3000\n\n")
    flusher.Flush()

    for {
        select {
        case <-c.done:
            return
        case e, ok := <-c.send:
            if !ok {
                return // channel was closed by unregister
            }
            payload, _ := json.Marshal(e.Data)
            fmt.Fprintf(w, "id: %d\n", e.ID)
            fmt.Fprintf(w, "event: %s\n", e.Type)
            fmt.Fprintf(w, "data: %s\n\n", payload)
            flusher.Flush()
        }
    }
}

Key design decisions:

Second Implementation Angle: Structured Event Builder #Permalink to this section

In production you want consistent event formatting and log correlation. A small helper avoids fmt.Fprintf scatter across handlers:

package sse

import (
    "fmt"
    "io"
    "strings"
)

// Writer wraps an http.ResponseWriter + http.Flusher pair.
type Writer struct {
    w io.Writer
    f interface{ Flush() }
}

// NewWriter returns a Writer or an error if w is not flushable.
func NewWriter(w io.Writer, f interface{ Flush() }) *Writer {
    return &Writer{w: w, f: f}
}

// Send writes one SSE event and flushes immediately.
// id == 0 omits the id field.
func (sw *Writer) Send(id int64, eventType, data string) error {
    var b strings.Builder
    if id > 0 {
        fmt.Fprintf(&b, "id: %d\n", id)
    }
    if eventType != "" && eventType != "message" {
        fmt.Fprintf(&b, "event: %s\n", eventType)
    }
    // Encode multi-line data per spec: each \n becomes \ndata: 
    lines := strings.Split(data, "\n")
    for _, line := range lines {
        fmt.Fprintf(&b, "data: %s\n", line)
    }
    b.WriteString("\n")

    if _, err := fmt.Fprint(sw.w, b.String()); err != nil {
        return err
    }
    sw.f.Flush()
    return nil
}

// Comment sends an SSE comment (colon-prefixed line) for keep-alive.
func (sw *Writer) Comment(text string) {
    fmt.Fprintf(sw.w, ": %s\n\n", text)
    sw.f.Flush()
}

Keep-alive comments (: ping\n\n) are important: many load balancers close idle connections after 60 seconds. A 30-second comment heartbeat prevents this. See HTTP Keep-Alive & Connection Lifecycle for timeout tables.

Edge Cases & Network Interference #Permalink to this section

Proxy buffering #Permalink to this section

Nginx buffers proxy_pass responses by default. An SSE stream sits silently behind a reverse proxy until Nginx’s buffer fills. Mitigation is two-pronged:

location /events {
    proxy_pass         http://backend:8080;
    proxy_http_version 1.1;
    proxy_set_header   Connection "";          # enable keep-alive to upstream
    proxy_buffering    off;                    # disable response buffering
    proxy_cache        off;
    proxy_read_timeout 86400s;                 # 24 h — prevent upstream timeout
    proxy_send_timeout 86400s;
}

The Go handler should also set X-Accel-Buffering: no as a belt-and-suspenders: Nginx honours this header per-response regardless of the proxy_buffering directive’s global value.

CDN stripping #Permalink to this section

CDNs (Cloudflare, Fastly, CloudFront) may:

• Buffer until the response ends — disable SSE entirely unless you configure streaming/pass-through mode.
• Strip Connection: keep-alive from HTTP/1.1 upstream responses (correct per HTTP/2 spec but breaks SSE over HTTP/1.1).
• Impose idle timeouts (Cloudflare default: 100 s). Send a comment heartbeat well under that threshold.

Always test with curl -N (no buffering) against the CDN edge to confirm events arrive incrementally.

Chunked transfer encoding #Permalink to this section

Go’s net/http server automatically applies Transfer-Encoding: chunked when no Content-Length is set and the response is streamed. Do not fight it. Intermediate proxies that strip Transfer-Encoding: chunked will reassemble the body — defeating streaming. For deep details on chunk framing, see Buffer Management & Chunked Transfer Encoding.

HTTP/2 and gRPC #Permalink to this section

Under HTTP/2 (Go’s net/http supports it natively), Flush() still works but http.Flusher is implemented differently by the H2 transport. The Connection: keep-alive header is a no-op for H2 and should not be sent (Go’s HTTP/2 server silently ignores it). Use http2.ConfigureServer and rely on Go’s H2 implementation to multiplex streams.

Firewall and idle TCP drops #Permalink to this section

Stateful firewalls drop idle TCP flows after 300–900 seconds without activity. A comment heartbeat every 25–30 seconds is the safest universal guard. Log every heartbeat at DEBUG level to make silence observable.

Disconnect detection #Permalink to this section

r.Context().Done() fires when the connection closes, but only if the handler is blocked on a select. If Write returns an error (e.g., broken pipe), the context is not yet cancelled — check errors.Is(err, syscall.EPIPE) or simply check the error and return:

if _, err := fmt.Fprintf(w, "data: ...\n\n"); err != nil {
    slog.Info("client write error, closing stream", "err", err)
    return
}

Performance & Scale Considerations #Permalink to this section

Connection counts and goroutines #Permalink to this section

Each SSE connection occupies one goroutine (≈8 KB stack, growing as needed), one file descriptor, and one buffered channel. A 64-entry chan Event costs 64 × (size of Event) bytes per client. At 10,000 connections with a 128-byte event struct: ~80 MB for channels plus ~80 MB for goroutine stacks — well within a single Go process’s typical memory budget.

For connection pooling decisions, the binding constraint is usually file descriptors (ulimit -n) and OS TCP socket memory, not goroutines.

Channel buffer sizing #Permalink to this section

For high-frequency streams you should apply rate limiting and backpressure at the hub level — not only at the channel.

Hub contention #Permalink to this section

The sync.Mutex in the hub is held for the duration of the fan-out loop. If you have 50,000 clients and Broadcast is called 100 times per second, the mutex is held for the time to iterate 50,000 pointers plus 50,000 non-blocking channel sends. Measured on an M2 core: ~2 ms for 10,000 clients. For higher cardinality, shard the hub into N independent hubs (hash by topic or user-ID mod N).

Serialisation cost #Permalink to this section

JSON marshalling inside Broadcast is wasteful: marshal once before iterating clients, then send []byte on the channel:

type rawEvent struct {
    id      int64
    evtType string
    payload []byte // pre-marshalled JSON
}

This avoids N marshal calls per broadcast.

Memory footprint and GC #Permalink to this section

A long-lived SSE handler that accumulates goroutine stack frames over hours causes gradual heap growth. Use pprof goroutine and heap profiles to identify leaks. The most common cause is a goroutine blocked on a channel send with no corresponding receiver — i.e., an unregistered client whose send channel was never closed.

For idempotent event IDs use atomic counters, not sync.Mutex-wrapped counters, to reduce contention on monotonic ID generation.

Validation & Debugging #Permalink to this section

curl smoke test #Permalink to this section

# -N disables curl's own output buffering; -H sets SSE headers
curl -N \
  -H "Accept: text/event-stream" \
  -H "Cache-Control: no-cache" \
  http://localhost:8080/events

Expected output (events appear as they are pushed, not in a batch):

retry: 3000

id: 1
event: tick
data: {"ts":1718966405000}

id: 2
event: tick
data: {"ts":1718966410000}

If all events arrive at once, a proxy or middleware is buffering. Check X-Accel-Buffering and proxy_buffering.

Go pprof #Permalink to this section

# Import _ "net/http/pprof" in main, then:
go tool pprof http://localhost:8080/debug/pprof/goroutine
# List goroutines blocked on channel receive — should equal active clients

Structured logging #Permalink to this section

Log every lifecycle event with log/slog (Go 1.21+):

slog.Info("sse: client connected",
    "remote_addr", r.RemoteAddr,
    "user_agent", r.UserAgent(),
)
defer slog.Info("sse: client disconnected",
    "remote_addr", r.RemoteAddr,
    "duration_s", time.Since(start).Seconds(),
)

Track active connection count with an atomic.Int64 gauge and expose it on /metrics (Prometheus) or /debug/vars (expvar). Alert when the gauge drops to zero unexpectedly — it signals a server-side panic or crash.

DevTools network tab #Permalink to this section

In Chrome DevTools → Network → filter EventStream: each SSE response shows a EventStream sub-tab with decoded events. Confirm Content-Type: text/event-stream in the response headers and check that the status is 200 (not 204 or a redirect).

Production Checklist #Permalink to this section

• http.Flusher type-assertion guarded with a non-nil error response
• Content-Type: text/event-stream, Cache-Control: no-cache, X-Accel-Buffering: no headers set before first write
• retry: directive sent in the first flush (recommend 3000–5000 ms)
• Client disconnect handled via r.Context().Done() — no goroutine leak on disconnect
• Hub Broadcast uses non-blocking channel send with drop-and-log, not a blocking send
• Comment heartbeat goroutine or ticker every ≤30 s per connection
• Nginx/proxy proxy_buffering off and proxy_read_timeout ≥ connection lifetime
• ulimit -n and /proc/sys/net/core/somaxconn tuned for expected connection count
• pprof endpoint available for goroutine leak diagnosis in staging
• Active connection count exported as a metric; alert rule configured

Frequently Asked Questions #Permalink to this section

Related #Permalink to this section

• Backend Stream Generation & Connection Management
• Implementing SSE with Go Channels and http.Flusher
• Graceful Shutdown for Go SSE Servers
• Buffer Management & Chunked Transfer Encoding
• Rate Limiting & Backpressure Handling
• Redis Pub/Sub Fan-Out for SSE
• Connection Pooling for SSE Servers