Versions in this module Expand all Collapse all v0 v0.2.0 Oct 28, 2026 Changes in this version + func Merge[T any](sources ...iter.Seq[Event[T]]) iter.Seq[Event[T]] + func PercentileDelay(q float64, window int) func() DelayStrategy + func Pipe[T any](ctx context.Context, in <-chan Event[T], opts ...Option) <-chan Event[T] + func Reorder[T any](seq iter.Seq[Event[T]], opts ...Option) iter.Seq[Event[T]] + type Aligned struct + func NewAligned[S comparable](opts ...AlignedOption) *Aligned[S] + func (a *Aligned[S]) Idle(src S) (combined time.Time, advanced, ok bool) + func (a *Aligned[S]) Lag() []SourceStatus[S] + func (a *Aligned[S]) Len() int + func (a *Aligned[S]) Observe(src S, wm time.Time) (combined time.Time, advanced, ok bool) + func (a *Aligned[S]) Register(src S) + func (a *Aligned[S]) Remove(src S) (combined time.Time, advanced, ok bool) + func (a *Aligned[S]) Watermark() (time.Time, bool) + type AlignedOption func(*alignedOptions) + func WithAlignedClock(now func() time.Time) AlignedOption + func WithSourceIdleTimeout(d time.Duration) AlignedOption + type Buffer struct + func New[T any](opts ...Option) *Buffer[T] + func (b *Buffer[T]) Add(t time.Time, v T) []Event[T] + func (b *Buffer[T]) AddTo(dst []Event[T], t time.Time, v T) []Event[T] + func (b *Buffer[T]) AdvanceTo(t time.Time) []Event[T] + func (b *Buffer[T]) Flush() []Event[T] + func (b *Buffer[T]) Len() int + func (b *Buffer[T]) OnLate(fn func(Event[T])) + func (b *Buffer[T]) Stats() Stats + func (b *Buffer[T]) Watermark() time.Time + type Dedup struct + func NewDedup[K comparable](window time.Duration, maxIDs int) *Dedup[K] + func (d *Dedup[K]) Duplicates() uint64 + func (d *Dedup[K]) Len() int + func (d *Dedup[K]) Seen(t time.Time, id K) bool + type DelayStrategy interface + Delay func() time.Duration + Observe func(lag time.Duration) + type Event struct + Late bool + Time time.Time + Value T + type Keyed struct + func NewKeyed[K comparable, T any](opts ...Option) *Keyed[K, T] + func (k *Keyed[K, T]) Add(key K, t time.Time, v T) []Event[T] + func (k *Keyed[K, T]) AdvanceAll(t time.Time, fn func(key K, events []Event[T])) + func (k *Keyed[K, T]) AdvanceTo(key K, t time.Time) []Event[T] + func (k *Keyed[K, T]) Flush(key K) []Event[T] + func (k *Keyed[K, T]) FlushAll(fn func(key K, events []Event[T])) + func (k *Keyed[K, T]) Len() int + func (k *Keyed[K, T]) OnLate(fn func(key K, e Event[T])) + func (k *Keyed[K, T]) Stats(key K) (s Stats, ok bool) + type KeyedTracker struct + func NewKeyedTracker[K comparable]() *KeyedTracker[K] + func (k *KeyedTracker[K]) Commit(key K) (offset int64, ok bool) + func (k *KeyedTracker[K]) Done(key K, offset int64) bool + func (k *KeyedTracker[K]) Len() int + func (k *KeyedTracker[K]) Revoke(key K) (offset int64, ok bool) + func (k *KeyedTracker[K]) Track(key K, offset int64) bool + type LatePolicy int + const Drop + const Emit + type Option func(*options) + func WithDelayStrategy(newStrategy func() DelayStrategy) Option + func WithIdleTimeout(d time.Duration) Option + func WithLatePolicy(p LatePolicy) Option + func WithMaxBuffered(n int) Option + func WithMaxDelay(d time.Duration) Option + type Record struct + Offset int64 + Time time.Time + Value T + type Safe struct + func NewSafe[T any](opts ...Option) *Safe[T] + func (s *Safe[T]) Add(t time.Time, v T) []Event[T] + func (s *Safe[T]) AddTo(dst []Event[T], t time.Time, v T) []Event[T] + func (s *Safe[T]) AdvanceTo(t time.Time) []Event[T] + func (s *Safe[T]) Flush() []Event[T] + func (s *Safe[T]) OnLate(fn func(Event[T])) + func (s *Safe[T]) Stats() Stats + type SafeKeyed struct + func NewSafeKeyed[K comparable, T any](opts ...Option) *SafeKeyed[K, T] + func (s *SafeKeyed[K, T]) Add(key K, t time.Time, v T) []Event[T] + func (s *SafeKeyed[K, T]) AdvanceAll(t time.Time, fn func(key K, events []Event[T])) + func (s *SafeKeyed[K, T]) AdvanceTo(key K, t time.Time) []Event[T] + func (s *SafeKeyed[K, T]) Flush(key K) []Event[T] + func (s *SafeKeyed[K, T]) FlushAll(fn func(key K, events []Event[T])) + func (s *SafeKeyed[K, T]) Len() int + func (s *SafeKeyed[K, T]) OnLate(fn func(key K, e Event[T])) + func (s *SafeKeyed[K, T]) Stats(key K) (Stats, bool) + type SourceStatus struct + Behind time.Duration + Holding bool + Idle bool + Reported bool + Source S + Watermark time.Time + type Stats struct + Buffered int + Late uint64 + Peak int + Released uint64 + Watermark time.Time + type Stream struct + func NewStream[K comparable, T any](opts ...StreamOption) *Stream[K, T] + func (s *Stream[K, T]) AdvanceAll(t time.Time, fn func(key K, records []Record[T])) + func (s *Stream[K, T]) Commit(key K) (offset int64, ok bool) + func (s *Stream[K, T]) Done(key K, offset int64) bool + func (s *Stream[K, T]) Flush(fn func(key K, r Record[T])) + func (s *Stream[K, T]) Ingest(key K, offset int64, t time.Time, id string, v T) []Record[T] + func (s *Stream[K, T]) OnLate(fn func(key K, r Record[T])) + func (s *Stream[K, T]) Revoke(key K, fn func(Record[T])) (offset int64, ok bool) + func (s *Stream[K, T]) Skip(key K, offset int64) bool + func (s *Stream[K, T]) Stats() StreamStats + func (s *Stream[K, T]) Watermark(key K) (time.Time, bool) + type StreamOption func(*streamConfig) + func WithBuffer(opts ...Option) StreamOption + func WithDedup(window time.Duration, maxIDs int) StreamOption + type StreamStats struct + Duplicates uint64 + Keys int + Late uint64 + Skipped uint64 + type Tracker struct + func NewTracker() *Tracker + func (t *Tracker) Commit() (offset int64, ok bool) + func (t *Tracker) Done(offset int64) bool + func (t *Tracker) Pending() int + func (t *Tracker) Track(offset int64) bool
Go to main page