Chapter 24 — Streaming over HTTP: Server-Sent Events, WebSockets, and long-lived connections

Build it in Go (30 min) — SSE, and the WriteTimeout trap

sse serves two endpoints from one server with WriteTimeout: 3s, a sensible value for ordinary requests. /events-naive ignores the timeout; /events extends its own write deadline before each event with http.ResponseController (Go 1.20+):

// sse: Server-Sent Events -- a long-lived HTTP response that streams events --
// and the production details that break it: flushing, heartbeats, resuming
// with Last-Event-ID, and a server WriteTimeout that silently cuts streams.
//
//	go run ./sse                 # runs the server and two clients, prints a timeline
//	go run ./sse -serve :8090    # then: curl -N localhost:8090/events
package main

import (
	"bufio"
	"flag"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"time"
)

// events streams one event per second. If extend is true, it pushes its own
// write deadline forward before every write, so a server-wide WriteTimeout
// (meant for normal requests) doesn't kill this long-lived stream.
func events(extend bool) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		rc := http.NewResponseController(w) // Go 1.20+: Flush, SetWriteDeadline per request
		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Cache-Control", "no-cache")
		w.Header().Set("X-Accel-Buffering", "no") // tells nginx not to buffer this response

		id := 0
		if last := r.Header.Get("Last-Event-ID"); last != "" { // a reconnecting client resumes
			id, _ = strconv.Atoi(last)
		}
		tick := time.NewTicker(time.Second)
		defer tick.Stop()
		for {
			select {
			case <-r.Context().Done(): // client went away: stop working for it
				return
			case <-tick.C:
				if extend {
					rc.SetWriteDeadline(time.Now().Add(10 * time.Second))
				}
				id++
				if id%5 == 0 {
					fmt.Fprint(w, ": heartbeat\n\n") // a comment line: keeps proxies/NATs from idling us out
				}
				_, err := fmt.Fprintf(w, "id: %d\nevent: tick\ndata: {\"n\":%d}\n\n", id, id)
				if err == nil {
					err = rc.Flush() // without this, events sit in a buffer
				}
				if err != nil {
					return
				}
			}
		}
	}
}

func main() {
	serve := flag.String("serve", "", "just run the server on this address")
	flag.Parse()
	mux := http.NewServeMux()
	mux.Handle("/events", events(true))
	mux.Handle("/events-naive", events(false))
	srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second,
		WriteTimeout: 3 * time.Second} // a "sensible" timeout for normal requests...
	if *serve != "" {
		srv.Addr = *serve
		fmt.Println("serving SSE on", *serve, "-- try: curl -N localhost"+*serve+"/events")
		srv.ListenAndServe()
		return
	}

	ts := httptest.NewUnstartedServer(mux)
	ts.Config = srv
	ts.Start()
	defer ts.Close()

	for _, path := range []string{"/events-naive", "/events"} {
		fmt.Printf("== %s (server WriteTimeout = 3s)\n", path)
		listen(ts.URL+path, "", 6*time.Second)
	}
	fmt.Println("== /events again, resuming with Last-Event-ID: 4")
	listen(ts.URL+"/events", "4", 2500*time.Millisecond)
}

func listen(url, lastID string, d time.Duration) {
	req, _ := http.NewRequest("GET", url, nil)
	if lastID != "" {
		req.Header.Set("Last-Event-ID", lastID)
	}
	start := time.Now()
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		fmt.Println(err)
		return
	}
	defer resp.Body.Close()
	sc := bufio.NewScanner(resp.Body)
	done := time.After(d - time.Since(start)) // d counts from the request, not the headers
	lines := make(chan string)
	go func() {
		for sc.Scan() {
			lines <- sc.Text()
		}
		close(lines)
	}()
	for {
		select {
		case <-done:
			fmt.Printf("  %4.1fs  (client stops listening)\n\n", time.Since(start).Seconds())
			return
		case l, ok := <-lines:
			if !ok {
				fmt.Printf("  %4.1fs  STREAM CUT by the server (%v)\n\n", time.Since(start).Seconds(), sc.Err())
				return
			}
			if strings.HasPrefix(l, "data:") || strings.HasPrefix(l, ":") {
				fmt.Printf("  %4.1fs  %s\n", time.Since(start).Seconds(), l)
			}
		}
	}
}

$ go run ./sse
== /events-naive (server WriteTimeout = 3s)
   1.0s  data: {"n":1}
   2.0s  data: {"n":2}
   3.0s  STREAM CUT by the server (unexpected EOF)

== /events (server WriteTimeout = 3s)
   1.0s  data: {"n":1}
   2.0s  data: {"n":2}
   3.0s  data: {"n":3}
   4.0s  data: {"n":4}
   5.0s  : heartbeat
   5.0s  data: {"n":5}
   6.0s  (client stops listening)

== /events again, resuming with Last-Event-ID: 4
   1.0s  : heartbeat
   1.0s  data: {"n":5}
   2.0s  data: {"n":6}
   2.5s  (client stops listening)

What to notice:

• The naive stream died at exactly 3.0 s. WriteTimeout covers the whole response, and for a stream "the whole response" never ends. This is one of the most common Go streaming bugs, and it looks like a network problem.
• rc.SetWriteDeadline extends the deadline per request, so normal endpoints keep the protective 3 s timeout and only the stream gets more time.
• Resume works: a client reconnecting with Last-Event-ID: 4 continues at event 5. Browsers' EventSource sends this header automatically when they reconnect.
• r.Context().Done() fires when the client disconnects, so the handler stops instead of writing to nobody.

Exercises:

1. Put Chapter 20's revproxy in front of the SSE server. Do events still arrive one per second? (They should: Go flushes text/event-stream.) Then put nginx in front with default settings and compare.
2. Add a broadcast hub: one goroutine produces events, and each connected client gets a buffered channel. Decide what happens when a client's channel is full (drop the client? drop events?). That decision is the core of every real-time system.
3. On SIGTERM, send each client a final event: reconnect with a retry: field set to a random delay, so they don't all come back at once.