Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var PackageProps = packageprops.Props{ WASMWASI: packageprops.WASMCompiles, WASMJS: packageprops.WASMCompiles, WASMFreestanding: packageprops.WASMCompiles, }
PackageProps records this package's curated properties (ADR-0080). Seeded by `wasmsurvey props generate`; curate by hand, then `wasmsurvey props verify`.
Functions ¶
This section is empty.
Types ¶
type BatchReaderI ¶
type BatchReaderI[K comparable, V EntityItem[K]] interface { StreamBatches(ctx context.Context) iter.Seq2[[]V, error] }
func Prefetcher ¶
func Prefetcher[K comparable, V EntityItem[K]]( source BatchReaderI[K, V], depth int, ) BatchReaderI[K, V]
Prefetcher wraps a BatchReaderI and reads 'depth' batches ahead.
The producer goroutine is bound to the lifetime of each StreamBatches call: when the consumer stops iterating (via yield-false or normal completion), an internal sub-context is cancelled so the producer exits even if the outer ctx is still live. The upstream source must honor its ctx for this to terminate promptly.
type ChunkPoolI ¶
type ChunkPoolI[T any] interface { Get() []T Put([]T) }
ChunkPoolI defines the contract for memory pooling.
type Config ¶
type Config struct {
BufferSize int // Channel buffer size
ChunkPoolCap int // Capacity of pooled slices
}
func DefaultConfig ¶
func DefaultConfig() Config
type ConsumerI ¶
type ConsumerI[K comparable, V EntityItem[K]] interface { Process(ctx context.Context, id K, rows iter.Seq[V]) (err error) }
type EntityItem ¶
type EntityItem[K comparable] interface { GetEntityID() K }
type MetricsCollectorI ¶
type MetricsCollectorI interface {
RecordBatch() // a non-empty batch was consumed from the source
RecordRows(n int) // n rows were forwarded to the active consumer
RecordEntityFinalized(ok bool) // ok=true if Process returned nil; false on error or panic
RecordEntityDuration(d time.Duration) // wall time from goroutine spawn to consumer return
}
MetricsCollectorI defines the observability hooks fired by Processor.Run.
All hooks are called from Run's own goroutine (the caller's goroutine), so the collector need not be goroutine-safe for sequential use. Callers that share one collector across concurrent Run invocations are responsible for the collector's thread safety.
type Option ¶
type Option[K comparable, V EntityItem[K]] func(*Processor[K, V])
Option mutates a Processor at construction time. See WithPool.
func WithMetrics ¶
func WithMetrics[K comparable, V EntityItem[K]](m MetricsCollectorI) Option[K, V]
WithMetrics installs an observability collector. If not set, a no-op collector is used so the Run loop can fire hooks without nil checks.
func WithPool ¶
func WithPool[K comparable, V EntityItem[K]](pool ChunkPoolI[V]) Option[K, V]
WithPool overrides the default chunk pool. Useful for tests that want to observe Get/Put behavior, or for callers that want to share a pool across processors.
type Processor ¶
type Processor[K comparable, V EntityItem[K]] struct { // contains filtered or unexported fields }
func NewProcessor ¶
func NewProcessor[K comparable, V EntityItem[K]]( consumer ConsumerI[K, V], cfg Config, opts ...Option[K, V], ) *Processor[K, V]
NewProcessor constructs a Processor. BufferSize and ChunkPoolCap must be non-negative; negative values panic at construction time rather than causing confusing make-time crashes later. BufferSize == 0 is valid (unbuffered handoff between reader and consumer); ChunkPoolCap == 0 is valid (no preallocation, append grows from nil).
func (*Processor[K, V]) Run ¶
func (inst *Processor[K, V]) Run(ctx context.Context, source BatchReaderI[K, V]) (err error)
Run streams batches from source, partitions rows by entity ID, and invokes the consumer on each entity's row stream in a dedicated goroutine.
Consumer contract: Process MUST honor ctx. On cancellation Run closes the row channel and waits for the consumer goroutine to exit before returning, so a consumer that ignores ctx will block Run indefinitely.
A consumer that returns nil before consuming all of its rows signals "done with this entity"; remaining rows for that entity are dropped and Run continues with the next entity. A non-nil return aborts the pipeline; the error is wrapped as "consumer for entity <id>: <err>" so the failing entity is identifiable.
Rows are assumed grouped by entity ID; a non-contiguous reappearance of the same ID is treated as a new lifecycle (a fresh consumer goroutine).
Run may be called multiple times on the same Processor — the chunk pool is reusable. Concurrent Run calls on the same Processor share the supplied consumer; if you call Run from multiple goroutines, the consumer must be safe to invoke from multiple goroutines.
Observability hooks (see WithMetrics) fire from Run's own goroutine in the order: RecordBatch (per non-empty batch), RecordRows (per chunk sent to a consumer), RecordEntityFinalized + RecordEntityDuration (once per entity lifecycle, regardless of success or failure).
type SlicePool ¶
type SlicePool[T any] struct { // contains filtered or unexported fields }
SlicePool implements a type-safe sync.Pool for slices.
func NewSlicePool ¶
func NewSlicePool[T any](capacity int, opts ...SlicePoolOption[T]) *SlicePool[T]
type SlicePoolOption ¶
SlicePoolOption mutates a SlicePool at construction time.
func WithZeroOnPut ¶
func WithZeroOnPut[T any]() SlicePoolOption[T]
WithZeroOnPut configures the pool to clear slice elements before returning the slice to the pool. Use this when T contains pointers (or string / slice / interface / map values) and you want the references to be released for GC instead of being retained in the pool's reused backing array.
Source Files