When you consume a server-sent events feed from an LLM API, you quickly discover that go partial json chunks sse handling is not as simple as calling json.Unmarshal on each line. The stream emits fragments of a JSON document across multiple events, and a naive line-by-line decode fails on the first incomplete payload. This guide walks through a concrete pattern: read SSE frames, push raw data onto a channel, buffer until a complete JSON value is present, then decode and dispatch using goroutines.
Step 1: Build a minimal SSE reader
Start with an HTTP request that sets Accept: text/event-stream. Use a bufio.Scanner with an enlarged buffer because the default is 64 KB and some JSON deltas exceed that. If you point this at an OpenAI-compatible endpoint such as n4n.ai, each event is usually a complete chat completion delta, but the same buffering logic protects you when a proxy splits the frame mid-object.
req, _ := http.NewRequest("GET", url, nil)
req.Header.Set("Accept", "text/event-stream")
req.Header.Set("Cache-Control", "no-cache")
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
scanner := bufio.NewScanner(resp.Body)
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)Iterate lines and skip everything that does not start with data: . SSE allows multi-line data fields; for simplicity assume single-line events, which is what most LLM gateways emit. The scan loop must stay thin—no parsing, no blocking work.
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
payload := strings.TrimPrefix(line, "data: ")
if payload == "[DONE]" {
break
}
// forward payload to processing goroutine
}
if err := scanner.Err(); err != nil {
log.Printf("scan error: %v", err)
}Step 2: Push raw chunks into a channel
Do not parse inside the scan loop. Hand the raw bytes to a channel and let a separate goroutine own the buffer. This keeps the read loop fast and avoids blocking on slow consumers. The channel buffer size of 16 absorbs brief stalls; if the consumer falls behind, the goroutine blocks, which is correct backpressure.
chunkCh := make(chan []byte, 16)
go func() {
defer close(chunkCh)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
payload := strings.TrimPrefix(line, "data: ")
if payload == "[DONE]" {
return
}
// copy because scanner reuses the underlying array
b := make([]byte, len(payload))
copy(b, payload)
chunkCh <- b
}
}()Copying the slice is mandatory: scanner.Text() returns a string, but converting to []byte without copy can alias the scanner’s internal buffer and corrupt data once the next scan completes.
Step 3: Accumulate bytes and detect complete JSON
A JSON value is complete only when json.Valid returns true. Partial chunks—like {"choices":[{"delta":{"con—fail that check. Maintain a bytes.Buffer, append incoming chunks, and use json.Decoder to consume exactly one value at a time. The decoder’s InputOffset tells you how many bytes were consumed so you can trim the buffer.
var buf bytes.Buffer
for chunk := range chunkCh {
buf.Write(chunk)
for {
if buf.Len() == 0 {
break
}
dec := json.NewDecoder(&buf)
var raw json.RawMessage
if err := dec.Decode(&raw); err != nil {
break // incomplete JSON, wait for more chunks
}
// raw holds one complete JSON value
workCh <- raw
// discard consumed bytes, keep leftovers
rest := buf.Bytes()[dec.InputOffset():]
buf.Reset()
buf.Write(rest)
}
}
close(workCh)This loop cleanly handles go partial json chunks sse where multiple JSON objects arrive concatenated or split arbitrarily across SSE events. If the buffer grows without ever becoming valid JSON, add a guard: if buf.Len() > 1<<20 { log.Fatal("unbounded partial json") }.
Step 4: Parse the complete object in a worker goroutine
The handle function runs concurrently. Keep it isolated from I/O. Define a struct matching the expected shape from the stream.
type Delta struct {
Choices []struct {
Delta struct {
Content string `json:"content"`
} `json:"delta"`
} `json:"choices"`
}
func handle(raw json.RawMessage) {
var d Delta
if err := json.Unmarshal(raw, &d); err != nil {
log.Printf("schema mismatch: %v", err)
return
}
for _, c := range d.Choices {
if c.Delta.Content != "" {
fmt.Print(c.Delta.Content)
}
}
}Spawning a goroutine per object is fine for low-frequency SSE (a few hundred per second). For higher throughput, use a worker pool fed by a channel instead of go handle(raw) to avoid goroutine explosions.
Step 5: Use a bounded worker pool
Replace ad-hoc goroutines with a fixed set of workers. This caps memory and CPU under bursty streams.
workCh := make(chan json.RawMessage, 64)
var wg sync.WaitGroup
for i := 0; i < 4; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for raw := range workCh {
handle(raw)
}
}()
}In the decode loop from Step 3, send with workCh <- raw. After chunkCh drains and workCh is closed, call wg.Wait() to exit cleanly. The pipeline now has three stages: network scan, buffer/decode, and parallel handle.
Step 6: Write a mock SSE server to verify
You cannot verify streaming code without a stream. Stand up a trivial HTTP server that splits a JSON object across two events to prove your buffering works.
func mockSSE(w http.ResponseWriter, r *http.Request) {
flusher, _ := w.(http.Flusher)
w.Header().Set("Content-Type", "text/event-stream")
parts := []string{
`{"choices":[{"delta":{"content":"Hel"`,
`lo"}}]}`,
}
for _, p := range parts {
fmt.Fprintf(w, "data: %s\n\n", p)
flusher.Flush()
time.Sleep(50 * time.Millisecond)
}
fmt.Fprint(w, "data: [DONE]\n\n")
}Run the client against this server using httptest.NewServer. If your buffer logic is correct, the two fragments concatenate into {"choices":[{"delta":{"content":"Hello"}}]} and the client prints Hello. A naive json.Unmarshal on the first chunk would error; the test proves the fix.
Step 7: Handle connection drops and cleanup
SSE is long-lived. Add a context to cancel the read goroutine and close channels on shutdown.
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() {
for {
select {
case <-ctx.Done():
return
case chunk, ok := <-chunkCh:
if !ok {
return
}
buf.Write(chunk)
// ... decode loop
}
}
}()Also check scanner.Err() after the loop; a broken pipe mid-stream is common. Log it and reconnect with a Last-Event-ID if the server supports it. Per-event metadata such as provider cache-control hints or routing directives should be extracted in the scan loop before the payload hits the channel—they are not part of the JSON body.
Verify success
Your test should assert these outcomes:
- The client prints the fully assembled string from split chunks with no
invalid characterpanics. - Backpressure works: inject a 100 ms sleep in
handleand confirm the scanner goroutine blocks rather than spawning unbounded goroutines. - A malformed final chunk (e.g.,
{"bad":) does not crash the process; the buffer guard logs and resets.
Run go test with the mock server. Assert that the collected content equals Hello. That confirms your go partial json chunks sse pipeline is solid.
Why this shape
The separation of scan → channel → buffer/decoder → worker pool matches Go’s concurrency model. You never block the network read on JSON parsing, and you never parse partial bytes prematurely. The json.Decoder.InputOffset trick avoids writing a custom JSON state machine, which is error-prone and easy to get wrong on unicode escapes. If you later forward these events to another system, do it inside the worker pool so the stream reader stays focused on bytes.
That is the whole pattern. Copy the loops, adjust the struct to your schema, and your streaming integration will survive fragmented frames and rate-limited providers without silent data loss.