Streaming LLM outputs arrive as a single sequential token stream, yet production systems often need that same stream consumed by loggers, metrics collectors, and client-facing websockets simultaneously. A go fan-out goroutines llm streaming architecture solves this by multiplexing one source onto many independent workers using channels, keeping the producer unblocked and the consumers isolated.
Step 1: Establish a streaming HTTP client
Use the standard net/http client to call an OpenAI-compatible chat completions endpoint with stream: true. The response body is server-sent events (SSE); each data: line carries a JSON delta. If you point this at n4n.ai, you get one endpoint that fronts 240+ models and automatic fallback, but the parsing code below is identical for any compliant gateway.
Set explicit timeouts on the client. LLM streams can stall; a context.WithTimeout at the call site prevents a hung connection from leaking goroutines.
package main
import (
"bytes"
"context"
"encoding/json"
"errors"
"net/http"
"time"
)
type chatReq struct {
Model string `json:"model"`
Messages []struct {
Role string `json:"role"`
Content string `json:"content"`
} `json:"messages"`
Stream bool `json:"stream"`
}
func openStream(ctx context.Context, endpoint, apiKey, model, prompt string) (*http.Response, error) {
reqBody := chatReq{
Model: model,
Stream: true,
Messages: []struct{ Role, Content string }{{"user", prompt}},
}
b, _ := json.Marshal(reqBody)
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(b))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+apiKey)
// Example of forwarding provider cache-control hint
req.Header.Set("Cache-Control", "max-age=300")
client := &http.Client{Timeout: 30 * time.Second}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
return nil, errors.New("bad status: " + resp.Status)
}
return resp, nil
}This returns the live body. Do not buffer it; pass resp.Body directly to the parser.
Step 2: Parse SSE into a typed channel
Wrap the body in a bufio.Scanner and extract the data: payloads. SSE lines can exceed the default 64KB scanner buffer for large tool-call chunks, so increase it. Skip the [DONE] sentinel. Push each decoded chunk into a chan CompletionChunk. Closing the channel signals all downstream consumers that the stream ended.
type CompletionChunk struct {
Choices []struct {
Delta struct {
Content string `json:"content"`
} `json:"delta"`
} `json:"choices"`
}
func parseStream(ctx context.Context, body io.Reader, out chan<- CompletionChunk) error {
defer close(out)
scanner := bufio.NewScanner(body)
scanner.Buffer(make([]byte, 0, 1024*1024), 4*1024*1024)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data:") {
continue
}
payload := strings.TrimSpace(line[len("data:"):])
if payload == "[DONE]" {
return nil
}
var chunk CompletionChunk
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
return err
}
select {
case out <- chunk:
case <-ctx.Done():
return ctx.Err()
}
}
return scanner.Err()
}Gateways like n4n.ai emit per-token usage metering in trailing headers; capture resp.Trailer after the loop if you need accurate cost attribution. The select with ctx.Done() ensures that if the consumer side cancels, the parser does not leak.
Step 3: Build the fan-out dispatcher
Fan-out means one input channel feeds N output channels. A common mistake is to share a single channel among multiple receivers—that is load-balancing, not fan-out. Each worker must receive every chunk. Create one buffered channel per consumer and a goroutine that copies each source chunk to all outputs.
func fanOut(src <-chan CompletionChunk, consumers int, bufSize int) []chan CompletionChunk {
outs := make([]chan CompletionChunk, consumers)
for i := range outs {
outs[i] = make(chan CompletionChunk, bufSize)
}
go func() {
for chunk := range src {
for _, out := range outs {
out <- chunk // blocks if any buffer full
}
}
for _, out := range outs {
close(out)
}
}()
return outs
}This naive copy blocks if any single consumer’s buffer is full, coupling worker speeds. For many workers, that coupling is unacceptable. In practice, use a non-blocking send with select and a drop policy, or per-worker goroutines that drain independently (see Step 4).
Step 4: Spawn worker goroutines with independent drains
Better than a central copy loop: each worker owns a goroutine that ranges over its channel. The dispatcher only does non-blocking broadcasts. Below, we launch three workers: a logger, a metrics counter, and a websocket forwarder stub.
func startWorkers(ctx context.Context, outs []chan CompletionChunk) *sync.WaitGroup {
var wg sync.WaitGroup
for i, out := range outs {
wg.Add(1)
go func(id int, ch <-chan CompletionChunk) {
defer wg.Done()
for chunk := range ch {
switch id {
case 0:
log.Printf("tok: %s", chunk.Choices[0].Delta.Content)
case 1:
metrics.Add(int64(len(chunk.Choices[0].Delta.Content)))
case 2:
// forward to ws conn; omitted for brevity
}
}
}(i, out)
}
return &wg
}If a worker is slow, its channel buffer fills and the dispatcher’s non-blocking send drops or skips. For critical consumers (like persistence), use unbounded buffering via a goroutine that appends to a slice, or increase bufSize. The go fan-out goroutines llm streaming pattern lives or dies on this backpressure decision.
Non-blocking broadcast variant
func broadcast(src <-chan CompletionChunk, outs []chan CompletionChunk, drops *int64) {
for chunk := range src {
for _, out := range outs {
select {
case out <- chunk:
default:
atomic.AddInt64(drops, 1)
}
}
}
}Use default only when loss is acceptable (e.g., live typing indicator). For billing or audit, block.
Step 5: Wire context cancellation and cleanup
A forgotten stream goroutine leaks file descriptors. Tie everything to a context.WithCancel rooted at the request. When the client disconnects, cancel, which unblocks parseStream and closes channels. Wait for workers with wg.Wait() before returning.
func runPipeline(ctx context.Context, endpoint, key, model, prompt string) error {
resp, err := openStream(ctx, endpoint, key, model, prompt)
if err != nil {
return err
}
defer resp.Body.Close()
src := make(chan CompletionChunk, 16)
go parseStream(ctx, resp.Body, src)
outs := fanOut(src, 3, 64)
wg := startWorkers(ctx, outs)
<-ctx.Done()
// fanOut closes outs when src closes; workers exit on closed chan
wg.Wait()
return nil
}If you use the non-blocking broadcast, call it instead of fanOut and ensure workers still exit on channel close. Always propagate the same ctx to parser and workers so a timeout cancels all in lockstep.
Step 6: Verify the pipeline end-to-end
Write a table-driven test that uses httptest to serve fake SSE, then count chunks received by a worker. This proves the go fan-out goroutines llm streaming logic without hitting a real model.
func TestFanOut(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
flusher := w.(http.Flusher)
w.Header().Set("Content-Type", "text/event-stream")
for i := 0; i < 3; i++ {
fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"x\"}}]}\n\n")
flusher.Flush()
}
fmt.Fprint(w, "data: [DONE]\n\n")
}))
defer srv.Close()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
resp, _ := openStream(ctx, srv.URL, "fake", "model", "hi")
defer resp.Body.Close()
src := make(chan CompletionChunk, 8)
go parseStream(ctx, resp.Body, src)
outs := fanOut(src, 2, 4)
var wg sync.WaitGroup
count := make([]int, 2)
for i, out := range outs {
wg.Add(1)
go func(id int, ch <-chan CompletionChunk) {
defer wg.Done()
for range ch {
count[id]++
}
}(i, out)
}
wg.Wait()
if count[0] != 3 || count[1] != 3 {
t.Fatalf("expected 3 chunks per worker, got %v", count)
}
}Run go test -run TestFanOut. If it passes, your fan-out correctly duplicates the stream. For production, add a metrics check that dropped chunks (if using non-blocking) stay within SLO, and assert that resp.Body is fully drained before close.
Practical notes
- Buffer sizes: start at 64; profile with real token rates (a 100 tok/s stream fills a 64-buffer in 0.6s if a worker stalls).
- Model routing: if you honor client routing directives, forward appropriate headers; the gateway forwards provider cache-control hints, so set
Cache-Controlon your request if you want prompt caching. - Never
rangeover a channel you haven’t closed; the dispatcher must close all outputs after source closes. - Use
errgroupfor cleaner cancellation if you prefer not to hand-roll WaitGroups. - Slow consumer isolation: a logger that writes to disk can block on fsync. Give it its own unbounded in-memory queue goroutine so it never stalls the broadcast loop.
The go fan-out goroutines llm streaming pattern is boring once wired correctly: one parser, N independent consumers, explicit backpressure. Get that right and adding a new side-effect (moderation, vector embed) is a single goroutine.