package sse
type Broker struct {
subscribers map[chan []byte]struct{}
subscribe chan chan []byte
unsubscribe chan chan []byte
publish chan []byte
}
func NewBroker() *Broker {
b := &Broker{
subscribers: make(map[chan []byte]struct{}),
subscribe: make(chan chan []byte),
unsubscribe: make(chan chan []byte),
publish: make(chan []byte, 16),
}
go b.run()
return b
}
func (b *Broker) run() {
for {
select {
case ch := <-b.subscribe:
b.subscribers[ch] = struct{}{}
case ch := <-b.unsubscribe:
if _, ok := b.subscribers[ch]; ok {
delete(b.subscribers, ch)
close(ch)
}
case msg := <-b.publish:
for ch := range b.subscribers {
select {
case ch <- msg:
default: // slow client: drop rather than block fan-out
}
}
}
}
}
func (b *Broker) Subscribe() chan []byte {
ch := make(chan []byte, 8)
b.subscribe <- ch
return ch
}
func (b *Broker) Unsubscribe(ch chan []byte) {
b.unsubscribe <- ch
}
func (b *Broker) Publish(msg []byte) {
b.publish <- msg
}package sse
import (
"fmt"
"net/http"
"time"
)
type Handler struct {
Broker *Broker
}
func (h *Handler) ServeHTTP(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")
ch := h.Broker.Subscribe()
defer h.Broker.Unsubscribe(ch)
keepalive := time.NewTicker(15 * time.Second)
defer keepalive.Stop()
ctx := r.Context()
for {
select {
case <-ctx.Done():
return
case <-keepalive.C:
fmt.Fprint(w, ": keepalive\n\n")
flusher.Flush()
case msg, open := <-ch:
if !open {
return
}
fmt.Fprintf(w, "data: %s\n\n", msg)
flusher.Flush()
}
}
}package main
import (
"fmt"
"log"
"net/http"
"time"
"example.com/app/sse"
)
func main() {
broker := sse.NewBroker()
go func() {
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for t := range ticker.C {
payload := fmt.Sprintf(`{"time":%q,"unix":%d}`, t.Format(time.RFC3339), t.Unix())
broker.Publish([]byte(payload))
}
}()
mux := http.NewServeMux()
mux.Handle("/events", &sse.Handler{Broker: broker})
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
fmt.Fprint(w, `<script>new EventSource('/events').onmessage=e=>console.log(e.data)</script>`)
})
log.Println("listening on :8080")
log.Fatal(http.ListenAndServe(":8080", mux))
}This snippet shows how to implement Server-Sent Events (SSE) in Go: a long-lived HTTP response that pushes a stream of text/event-stream messages to a browser without polling. SSE is a good fit for one-directional real-time updates (notifications, live metrics, progress bars) because it rides on plain HTTP, reconnects automatically in the browser, and is far simpler than WebSockets.
The Broker type in broker.go is a tiny in-memory pub/sub hub. It keeps a map of subscriber channels and runs a single event loop in run() that owns that map, so registration, removal, and fan-out all happen on one goroutine and no mutex is needed. Registration and deregistration flow through the subscribe and unsubscribe channels, while Publish pushes a []byte payload to every subscriber. The fan-out uses a select with a default branch: if a subscriber's buffered channel is full, the message is dropped for that client rather than blocking the whole broker. That is deliberate backpressure handling — one slow reader must never stall every other reader.
handler.go contains ServeHTTP, the actual SSE endpoint. It first type-asserts the http.ResponseWriter to an http.Flusher; without flushing, Go's buffering would hold bytes back and the client would see nothing until the connection closed. It sets the mandatory Content-Type: text/event-stream plus Cache-Control: no-cache and Connection: keep-alive headers, then registers a subscriber via broker.Subscribe() and guarantees cleanup with defer broker.Unsubscribe(ch).
The core loop selects over three things: r.Context().Done() fires when the browser disconnects, letting the goroutine exit cleanly instead of leaking; a time.Ticker emits a : keepalive comment line periodically so proxies don't drop an idle connection; and the subscriber channel delivers real events. Each event is written in the wire format data: <payload>\n\n and immediately followed by flusher.Flush() to push it down the socket.
main.go wires it together, starting the broker and firing a background goroutine that publishes JSON ticks. A key pitfall this design avoids is the goroutine leak: because every write path is guarded by context cancellation, closing the tab tears the whole chain down.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
