Last updated 2026-05-17

Package llmrouter — Streams

Every public symbol declared in stream.go. A Stream is the handle a consumer uses to read chunks; NewStream and ProducerHooks are the plumbing a provider implementation uses to push them.

Stream

The streaming-completion handle. Single-consumer: only one goroutine should read Chunks() for a given Stream. All fields are unexported; interact through the methods below. Source: stream.go#L10. 

type Stream struct {
    // unexported fields: chunks chan, cancel func, errMu chan, err error
}

Stream.Chunks

func (s *Stream) Chunks() <-chan Chunk

Returns the receive-only chunk channel. The channel is buffered with capacity 16 and closes exactly once, after the producer calls Finish. Source: stream.go#L29.

Once the channel is closed, any subsequent receive returns the zero Chunk immediately — the canonical for chunk := range stream.Chunks() loop terminates naturally.

Stream.Err

func (s *Stream) Err() error

Returns the terminal error, if any. Source: stream.go#L33.

Semantics:

  • Blocks until the producer finishes — internally it waits on a sentinel channel (errMu) that the producer closes inside Finish.
  • After unblocking, returns the same error value on every subsequent call.
  • Safe to call multiple times. Reading from a closed channel never blocks, so every reader gets the same answer.
  • Returns nil on clean stream completion. Returns context.Canceled / context.DeadlineExceeded if the context cancelled mid-stream. Returns the wrapped scanner error if reading the SSE body failed.

Stream.Cancel

func (s *Stream) Cancel()

Asks the producer to stop. Source: stream.go#L40.

Cancels the internal context derived from the parent that was passed to NewStream. The producer notices the cancellation on its next loop iteration, calls Finish(ctx.Err()), and closes Chunks(). Idempotent — calling cancel() multiple times is a no-op after the first.

Example: consumer-side stream lifecycle

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()


stream, err := provider.CompletionStream(ctx, req)
if err != nil {
    return err
}


// Optional: bail out after the first 200 characters.
var seen int
for chunk := range stream.Chunks() {
    for _, c := range chunk.Choices {
        seen += len(c.Delta.Content)
        fmt.Print(c.Delta.Content)
    }
    if seen >= 200 {
        stream.Cancel() // tell the producer to stop; loop drains until close
    }
}


if err := stream.Err(); err != nil && !errors.Is(err, context.Canceled) {
    return err
}

Stream lifecycle

The complete state machine for a single Stream:

  1. Provider calls NewStream(parent) — derived ctx, empty buffered channel (cap 16), unclosed errMu sentinel.
  2. Provider spawns a goroutine that calls hooks.Send(chunk) for each event from the upstream.
  3. Consumer ranges over stream.Chunks(), processing each chunk.
  4. Producer reaches end-of-stream ([DONE], message_stop) or hits a fatal error or notices ctx cancellation.
  5. Producer calls hooks.Finish(err). This closes the chunks channel and the errMu sentinel atomically.
  6. Consumer's range loop exits.
  7. Consumer calls stream.Err() — returns the value the producer passed to Finish.

hooks.Send returns false as soon as the internal context cancels (either because the parent context cancelled or because Stream.Cancel() was called). The producer is expected to react to false by calling Finish(ctx.Err()) and exiting.

NewStream

Exported for provider implementations in subpackages. Source: stream.go#L66. 

func NewStream(parent context.Context) (*Stream, context.Context, ProducerHooks)

Returns three things:

  1. The *Stream handed to the consumer (returned by the provider's CompletionStream).
  2. A derived context.Context the producer should respect for cancellation and request scoping.
  3. A ProducerHooks struct exposing Send and Finish.

Internally derives the context with context.WithCancel(parent) and stores the cancel function on the Stream so that Stream.Cancel can trip it.

ProducerHooks

The callbacks a provider uses to feed a Stream. Source: stream.go#L78. 

type ProducerHooks struct {
    Send   func(Chunk) bool
    Finish func(error)
}
Send(c Chunk) bool
Forwards one chunk to the consumer. Blocks if the buffered channel (cap 16) is full and the consumer is slow. Returns false as soon as the internal context cancels — the producer must then stop reading and call Finish(ctx.Err()).
Finish(err error)
Terminates the stream. Stores err as the terminal error, closes the chunks channel, and closes the internal errMu sentinel so any pending Stream.Err() calls unblock. Must be called exactly once per stream.

Example: implementing a provider

The full pattern used by both built-in providers. Send returns false on consumer cancellation; the producer bails out cleanly. Finish is called once in every exit path — including the error and cancel paths — so the consumer's Err() call never deadlocks. 

func (p *Provider) CompletionStream(ctx context.Context, req llmrouter.ChatRequest) (*llmrouter.Stream, error) {
    resp, err := p.openUpstream(ctx, req) // returns *http.Response, error
    if err != nil {
        return nil, err
    }
    if resp.StatusCode >= 400 {
        defer resp.Body.Close()
        body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
        return nil, &llmrouter.ErrUpstream{
            Provider:   p.Name(),
            StatusCode: resp.StatusCode,
            Body:       string(body),
        }
    }


    stream, sctx, hooks := llmrouter.NewStream(ctx)
    go func() {
        defer resp.Body.Close()
        scanner := bufio.NewScanner(resp.Body)
        for scanner.Scan() {
            if sctx.Err() != nil {
                hooks.Finish(sctx.Err())
                return
            }
            chunk, ok := parseLine(scanner.Text())
            if !ok {
                continue // skip malformed events
            }
            if !hooks.Send(chunk) {
                hooks.Finish(sctx.Err())
                return
            }
        }
        if err := scanner.Err(); err != nil {
            hooks.Finish(fmt.Errorf("read stream: %w", err))
            return
        }
        hooks.Finish(nil) // clean end-of-stream
    }()
    return stream, nil
}