stream

stream

Pull-based data pipeline with composable operators, plus a bounded push fan-out source

stream provides lazy, backpressure-aware data processing for Go. Pipelines pull values on demand — no work happens until you call Collect, Drain, or ForEach. This design naturally handles flow control without buffering or blocking. For the genuinely push-shaped "one source, many observers" case, Broadcaster fans events out to independent subscribers with bounded, drop-on-overflow buffers.

gokit converges on the rskit-stream operator vocabulary (map/filter/fan_out/window/batch/parallel/merge/partition/…) and its Broadcaster, expressed idiomatically in Go: a pull iterator for transformation pipelines and a bounded channel bus for fan-out. No operator buffers without bound.

Features

• Lazy evaluation — operators compose but don't execute until pulled
• Backpressure — upstream producers only generate values when downstream consumers request them
• Composable operators — map, filter, batch, throttle, window, and more
• Provider integration — structurally compatible with provider.Iterator[T]
• Context-aware — all operations support cancellation and deadlines
• Type-safe — full generic support for strongly typed pipelines

Install

go get github.com/kbukum/gokit@latest

Pipeline is part of the core module — no separate sub-module import needed.

Quick Start

package main

import (
    "context"
    "fmt"
    "github.com/kbukum/gokit/stream"
)

func main() {
    ctx := context.Background()
    
    // Create a pipeline from a slice
    src := stream.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10})
    
    // Double each value
    doubled := stream.Map(src, func(_ context.Context, n int) (int, error) {
        return n * 2, nil
    })
    
    // Keep only even numbers
    evens := stream.Filter(doubled, func(n int) bool {
        return n%2 == 0
    })
    
    // Collect results
    results, err := stream.Collect(ctx, evens)
    if err != nil {
        panic(err)
    }
    
    fmt.Println(results) // [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
}

Operators

Operator reference

Synchronous Operators

Concurrent Operators

Push Fan-Out (Broadcaster)

For the "watch one source → fan a typed change stream out to many independent observers" shape (config reloads, service discovery, cache invalidation, secret rotation), use Broadcaster[T]. Each subscriber owns a private bounded channel: a subscriber lagging beyond its buffer drops the overflow (backpressure by drop) but never blocks the broadcaster or its peers.

b := stream.NewBroadcaster[Config](stream.WithBroadcastBuffer(16))
defer b.Close()

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
updates := b.Subscribe(ctx)
go func() {
    for cfg := range updates { // ends when ctx is canceled or b.Close() is called
        reload(cfg)
    }
}()

b.Broadcast(newConfig) // reaches every live subscriber; slow ones drop overflow

Stream/Time-Aware Operators

Terminal Operators

Source Constructors

Usage Examples

Example 1: Data Transformation

ctx := context.Background()

// Process a list of user IDs
userIDs := stream.FromSlice([]string{"u1", "u2", "u3", "u4", "u5"})

// Fetch user data
users := stream.Map(userIDs, func(ctx context.Context, id string) (*User, error) {
    return userService.GetByID(ctx, id)
})

// Filter active users
active := stream.Filter(users, func(u *User) bool {
    return u.Active
})

// Collect results
activeUsers, err := stream.Collect(ctx, active)

Example 2: Batching & Throttling

// Stream of events
events := stream.From(eventSource) // eventSource implements Iterator[Event]

// Rate-limit to 10 events/sec
throttled := stream.Throttle(events, 100*time.Millisecond)

// Batch into groups of 50 or every 5 seconds
batched := stream.Batch(throttled, 50, 5*time.Second)

// Process each batch
stream.Drain(batched, func(ctx context.Context, batch []Event) error {
    return batchProcessor.Process(ctx, batch)
}).Run(ctx)

Example 3: Provider Integration

import (
    "github.com/kbukum/gokit/stream"
    "github.com/kbukum/gokit/provider"
)

// Assume audioSource implements provider.Iterator[AudioChunk]
src := stream.From(audioSource)

// Transcribe audio chunks
transcribed := stream.FlatMap(src, func(ctx context.Context, chunk AudioChunk) (Iterator[Segment], error) {
    return transcriber.Execute(ctx, chunk)
})

// Publish to Kafka as side-effect
tapped := stream.Tap(transcribed, func(ctx context.Context, seg Segment) error {
    return kafkaPublisher.Send(ctx, seg)
})

// Identify speakers
identified := stream.Map(tapped, func(ctx context.Context, seg Segment) (IdentifiedSegment, error) {
    return speakerID.Execute(ctx, seg)
})

// Drain to final sink
stream.Drain(identified, finalSink.Send).Run(ctx)

Example 4: Windowing

// Stream of metrics
metrics := stream.From(metricSource) // metricSource implements Iterator[Metric]

// Create 1-minute tumbling windows
windows := stream.TumblingWindow(metrics, 1*time.Minute)

// Aggregate each window
aggregated := stream.Map(windows, func(ctx context.Context, window []Metric) (Summary, error) {
    return aggregator.Summarize(window), nil
})

// Store summaries
stream.Drain(aggregated, summaryStore.Save).Run(ctx)

Example 5: FanOut (Parallel Processing)

// Process data through multiple models concurrently
src := stream.FromSlice(audioChunks)

results := stream.FanOut(src,
    modelA.Execute, // Run in parallel
    modelB.Execute,
    modelC.Execute,
)

// Each value is now []Result containing outputs from all 3 models
stream.Drain(results, func(ctx context.Context, outputs []Result) error {
    return combiner.Merge(ctx, outputs)
}).Run(ctx)

Example 6: Error Handling

src := stream.FromSlice(items)

processed := stream.Map(src, func(ctx context.Context, item Item) (Result, error) {
    // Errors propagate and halt the pipeline
    if err := validate(item); err != nil {
        return Result{}, fmt.Errorf("validation failed: %w", err)
    }
    return process(item), nil
})

// Collect will return the first error encountered
results, err := stream.Collect(ctx, processed)
if err != nil {
    log.Error("pipeline failed", map[string]interface{}{"error": err})
    return
}

Example 7: Parallel Processing with Workers

// Process 100 items with 10 concurrent workers
src := stream.FromSlice(items)

// Order is NOT preserved with Parallel
processed := stream.Parallel(src, 10, func(ctx context.Context, item Item) (Result, error) {
    return heavyProcessing(ctx, item)
})

results, err := stream.Collect(ctx, processed)

Design Philosophy

Pull vs Push

Pull-based (this library):

• Downstream pulls from upstream
• Natural backpressure — producer only works when consumer is ready
• No buffering required
• Lazy evaluation — nothing happens until terminal operator runs

Push-based (channels, reactive streams):

• Upstream pushes to downstream
• Requires explicit backpressure mechanism
• Buffering often needed
• Eager evaluation — producers start immediately

Structural Compatibility with Provider

The Iterator[T] interface is structurally identical to provider.Iterator[T]:

type Iterator[T any] interface {
    Next(ctx context.Context) (T, bool, error)
    Close() error
}

This means provider streams plug directly into pipelines:

// Provider iterator
audioIterator := audioProvider.Stream(ctx, input)

// Wrap in pipeline
p := stream.From(audioIterator)

// Apply operators
transcribed := stream.FlatMap(p, transcriber.Execute)

Testing

Use FromSlice for deterministic test data:

func TestPipeline(t *testing.T) {
    ctx := context.Background()
    src := stream.FromSlice([]int{1, 2, 3})
    
    doubled := stream.Map(src, func(_ context.Context, n int) (int, error) {
        return n * 2, nil
    })
    
    result, err := stream.Collect(ctx, doubled)
    require.NoError(t, err)
    assert.Equal(t, []int{2, 4, 6}, result)
}

Performance Considerations

• Lazy evaluation means no work until terminal operator runs
• Synchronous operators (Map, Filter) add negligible overhead
• Concurrent operators (Parallel, Buffer, Merge) spawn goroutines — use when I/O or CPU-heavy work benefits from parallelism
• Throttle/Batch/Debounce use timers — appropriate for real-time/streaming scenarios, not batch processing
• Context cancellation stops the pipeline immediately at the next operator

Related Packages

• provider — Provider pattern with Iterator interface (structurally compatible)
• dag — Dependency-ordered task orchestration with batch/stream modes
• sse — Server-sent events broadcasting (push-based, not pull-based)

License

MIT — Copyright (c) 2024 kbukum