Documentation
¶
Overview ¶
Package stream is the engine's substrate for large, pull-based results over interfaces.Stream.
FromSlice adapts an in-memory slice. Chunk splits items into ordered groups no larger than a byte budget, so one oversized result travels as several bounded frames. Collect drains a stream into a slice under a byte budget and fails with CodeResourceExhausted, closing the source, once the budget is spent — the guard a request/response surface puts in front of an unbounded stream. Buffer decouples a producer from its consumer: the bounded Kind reads ahead into a buffer capped by item count and bytes, blocking the producer when it is full (backpressure), and the direct Kind passes the stream through unchanged. NewFromConfig selects the Kind, and NewMetrics decorates a Buffer with read-ahead depth gauges and a producer-wait histogram.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Chunk ¶
Chunk splits items, in order, into consecutive groups whose sizes sum to at most maxBytes; size reports one item's encoded size. The split is greedy: a group closes only when the next item would not fit. An item larger than maxBytes on its own is CodeResourceExhausted, and a non-positive maxBytes is CodeInvalidInput. No items yield no groups.
func Collect ¶
func Collect[T any]( ctx context.Context, s interfaces.Stream[T], size func(T) int, maxBytes int, ) (items []T, err error)
Collect drains s into a slice and closes it. Once the items' sizes sum past maxBytes it stops reading and returns CodeResourceExhausted; a source error is returned unchanged; a non-positive maxBytes is CodeInvalidInput. s is closed on every path.
func FromSlice ¶
func FromSlice[T any](items []T) interfaces.Stream[T]
FromSlice returns a Stream that yields items in order and then ends. It does no I/O and never fails.
Types ¶
type Buffer ¶
type Buffer[T any] interface { // Wrap returns a stream yielding src's items in order, then src's end or error. // Closing the returned stream stops any producer Wrap started and closes src; ctx // bounds that producer's reads. Wrap(ctx context.Context, src interfaces.Stream[T]) interfaces.Stream[T] }
Buffer decouples a stream's producer from its consumer.
func NewFromConfig ¶
NewFromConfig builds the Buffer cfg selects, without I/O; size reports one item's size in bytes for KindBounded's byte bound. It runs once per item on the producer's path, so it should be cheap; a negative size counts as zero. An unknown kind, or KindBounded with a non-positive bound or a nil size, is CodeInvalidInput.
func NewMetrics ¶
func NewMetrics[T any]( inner Buffer[T], m interfaces.Metrics, name string, size func(T) int, ) Buffer[T]
NewMetrics wraps inner so every stream it buffers reports, under the label name, the items and bytes read from the source but not yet delivered (stream_buffer_items, stream_buffer_bytes) and the producer's wait between items (stream_buffer_producer_wait_seconds). size reports one item's size in bytes. The metrics are registered here, at construction; apply it at the composition root.
type Config ¶
type Config struct {
// Kind selects the backend; the zero value is KindDirect.
Kind Kind `yaml:"kind" mapstructure:"kind"`
// MaxItems caps the items KindBounded holds read-ahead; it must be positive for
// that kind.
MaxItems int `yaml:"max_items" mapstructure:"max_items"`
// MaxBytes caps the summed size of the items KindBounded holds read-ahead; it must
// be positive for that kind. One item larger than MaxBytes is still admitted, alone.
MaxBytes int `yaml:"max_bytes" mapstructure:"max_bytes"`
}
Config selects and bounds a Buffer.
Source Files