streaming-patterns_skill

Overview

This skill documents idiomatic streaming patterns for Go 1.23 using iter.Seq2[T, error], tailored for Beluga AI v2 streaming, transforms, and backpressure. It presents a primary primitive implementation, composition helpers, and concrete rules to keep streaming code safe, cancelable, and composable. Use it as a practical reference when implementing producers, transformers, or buffered streams in Go.

How this skill works

The core primitive is iter.Seq2[T, error], a function that accepts a yield callback to push items or an error to a consumer. Producers implement a loop that checks context cancellation, receives or generates chunks, and calls yield; yield returning false signals the consumer stopped and the producer must return. Composition helpers (Pipe, Collect, Fan-out, BufferedStream) wrap or coordinate Seq2 streams without exposing raw channels in the public API.

When to use it

• Implementing a network or model inference stream that must send data incrementally to a consumer.
• Transforming or mapping streamed items with backpressure awareness.
• Collecting streamed results into a slice or returning the last received item.
• Broadcasting a single producer to multiple consumers with fan-out semantics.
• Adding buffering to smooth rate mismatches while respecting cancellation and stop signals.

Best practices

• Expose iter.Seq2[T, error] in public APIs; avoid returning <-chan to callers.
• Always check ctx.Done() in producers and yield a context error if canceled.
• Respect yield returning false: stop work and return immediately to avoid leaks.
• Use internal channels and goroutines freely, but keep public boundaries as Seq2 functions.
• Reserve iter.Pull2 and explicit pull semantics for cases that truly need on-demand consumption.

Example use cases

• Stream model token chunks from a remote client and transform each chunk before yielding to the caller.
• Pipe a generator stream through a sanitizer transform that can return errors per item.
• Collect a stream into a slice to run a post-processing step on all chunks once the stream completes.
• Implement a buffered stream that decouples producer and consumer rates while propagating context cancellation.