Documentation
¶
Overview ¶
Package stream provides a generic streaming iterator for database results.
Stream[T] is a lazy, pull-based iterator over database results. It holds an open server-side cursor and decodes one row at a time. Supports Go 1.23+ range-over-func via the All method.
The streaming system supports composable transforms (Map, Filter, Reduce, Chunk, Take, ForEach) as zero-allocation pipeline operations.
Index ¶
- func ForEach[T any](s *Stream[T], fn func(T) error) error
- func Reduce[T, A any](s *Stream[T], initial A, fn func(A, T) A) (A, error)
- type BatchCursor
- type ChangeEvent
- type ChangeOp
- type ChangeSource
- type ChangeStream
- func (cs *ChangeStream[T]) All(yield func(ChangeEvent[T], error) bool)
- func (cs *ChangeStream[T]) Close() error
- func (cs *ChangeStream[T]) Err() error
- func (cs *ChangeStream[T]) Event() ChangeEvent[T]
- func (cs *ChangeStream[T]) Next(ctx context.Context) bool
- func (cs *ChangeStream[T]) ResumeToken() any
- type Cursor
- type DecodeFunc
- type HookRunner
- type Stream
- func Chunk[T any](s *Stream[T], size int) *Stream[[]T]
- func Filter[T any](s *Stream[T], fn func(T) bool) *Stream[T]
- func Map[T, U any](s *Stream[T], fn func(T) (U, error)) *Stream[U]
- func New[T any](cursor Cursor, decode DecodeFunc[T]) *Stream[T]
- func NewWithHooks[T any](cursor Cursor, decode DecodeFunc[T], hooks HookRunner, qc any) *Stream[T]
- func Take[T any](s *Stream[T], n int) *Stream[T]
- func (s *Stream[T]) All(yield func(T, error) bool)
- func (s *Stream[T]) Close() error
- func (s *Stream[T]) Collect(ctx context.Context) ([]T, error)
- func (s *Stream[T]) Count(ctx context.Context) (int64, error)
- func (s *Stream[T]) Err() error
- func (s *Stream[T]) Next(ctx context.Context) bool
- func (s *Stream[T]) Value() T
- func (s *Stream[T]) WithHooks(runner HookRunner, qc any) *Stream[T]
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type BatchCursor ¶
type BatchCursor struct {
// contains filtered or unexported fields
}
BatchCursor wraps a base cursor-creating function to implement batched fetch semantics. It fetches batchSize rows at a time from the underlying source, yielding them one at a time.
func NewBatchCursor ¶
func NewBatchCursor(fetchBatch func(offset, limit int) (Cursor, error), batchSize int) *BatchCursor
NewBatchCursor creates a BatchCursor that fetches rows in batches of batchSize. The fetchBatch function is called each time a new batch is needed, with the current offset and the batch size as limit.
func (*BatchCursor) Close ¶
func (bc *BatchCursor) Close() error
Close releases the current batch cursor and marks the BatchCursor as done.
func (*BatchCursor) Columns ¶
func (bc *BatchCursor) Columns() ([]string, error)
Columns returns the column names from the current batch's result set.
func (*BatchCursor) Err ¶
func (bc *BatchCursor) Err() error
Err returns any error encountered during iteration.
func (*BatchCursor) Next ¶
func (bc *BatchCursor) Next() bool
Next advances to the next row, fetching a new batch when the current batch is exhausted. Returns false when no more rows are available.
func (*BatchCursor) Scan ¶
func (bc *BatchCursor) Scan(dest ...any) error
Scan copies the current row's column values into dest.
type ChangeEvent ¶
type ChangeEvent[T any] struct { // Operation is the type of change. Operation ChangeOp // Before is the previous state (nil for inserts; available if driver supports it). Before *T // After is the new state (nil for deletes). After *T // Timestamp is the server timestamp of the change. Timestamp time.Time // ResumeToken is an opaque token for resuming the stream after disconnect. ResumeToken any }
ChangeEvent represents a single change from a CDC stream.
type ChangeSource ¶
type ChangeSource[T any] interface { // Next blocks until the next change event is available. // Returns false when the stream is closed or an error occurred. Next(ctx context.Context) bool // Event returns the current change event. Event() ChangeEvent[T] // Err returns any error encountered. Err() error // Close stops the change stream. Close() error // ResumeToken returns the current resume token for reconnection. ResumeToken() any }
ChangeSource is the interface that drivers implement to provide CDC events.
type ChangeStream ¶
type ChangeStream[T any] struct { // contains filtered or unexported fields }
ChangeStream is a long-lived iterator over real-time database changes. It wraps a ChangeSource and supports automatic reconnection with resume tokens.
func NewChangeStream ¶
func NewChangeStream[T any](source ChangeSource[T]) *ChangeStream[T]
NewChangeStream creates a new ChangeStream from a driver-specific source.
func (*ChangeStream[T]) All ¶
func (cs *ChangeStream[T]) All(yield func(ChangeEvent[T], error) bool)
All returns a range-over-func iterator for change events.
func (*ChangeStream[T]) Close ¶
func (cs *ChangeStream[T]) Close() error
Close stops the change stream and releases resources.
func (*ChangeStream[T]) Err ¶
func (cs *ChangeStream[T]) Err() error
Err returns any error encountered.
func (*ChangeStream[T]) Event ¶
func (cs *ChangeStream[T]) Event() ChangeEvent[T]
Event returns the current change event.
func (*ChangeStream[T]) Next ¶
func (cs *ChangeStream[T]) Next(ctx context.Context) bool
Next blocks until the next change event is available.
func (*ChangeStream[T]) ResumeToken ¶
func (cs *ChangeStream[T]) ResumeToken() any
ResumeToken returns the current resume token for reconnection.
type Cursor ¶
type Cursor interface {
// Next advances the cursor to the next row.
// Returns false when there are no more rows or an error occurred.
Next() bool
// Scan copies the current row's column values into dest.
Scan(dest ...any) error
// Columns returns the column names from the result set.
Columns() ([]string, error)
// Close releases the cursor and underlying resources.
Close() error
// Err returns any error encountered during iteration.
Err() error
}
Cursor is the interface that database drivers implement to provide streaming row access. Each driver wraps its native cursor type.
type DecodeFunc ¶
DecodeFunc decodes a cursor row into a value of type T.
type HookRunner ¶
type HookRunner interface {
RunStreamRowHook(ctx context.Context, qc any, row any) (int, error) // returns Decision as int
}
HookRunner is an optional interface for per-row hook execution. This avoids importing the hook package directly.
type Stream ¶
type Stream[T any] struct { // contains filtered or unexported fields }
Stream is a lazy, pull-based iterator over database results.
func New ¶
func New[T any](cursor Cursor, decode DecodeFunc[T]) *Stream[T]
New creates a new Stream from a cursor and a decode function.
func NewWithHooks ¶
func NewWithHooks[T any](cursor Cursor, decode DecodeFunc[T], hooks HookRunner, qc any) *Stream[T]
NewWithHooks creates a new Stream with per-row hook execution.
func (*Stream[T]) All ¶
All returns a range-over-func iterator for use with Go 1.23+ for-range.
for user, err := range userStream.All {
...
}
func (*Stream[T]) Close ¶
Close releases the server-side cursor and underlying connection. Always defer this.
func (*Stream[T]) Next ¶
Next advances the cursor and decodes the next row. Returns false when exhausted or on error (check Err()).
If hooks are configured, RunStreamRowHook is called after decoding each row. A Skip decision causes the row to be skipped (continue to next). A Deny decision stops iteration with an error.
Source Files