stream

func NewBroadcastHub ¶

func NewBroadcastHub[T any](ctx context.Context, src Stream[T], bufPerSubscriber int) *BroadcastHub[T]

NewBroadcastHub creates a BroadcastHub that reads from src and fans out every value and error to all current subscribers.

bufPerSubscriber is the buffer size for each subscriber's Values and Errors channels. A buffer of 0 is unbuffered — the hub blocks until the subscriber reads each item. Use a positive buffer to prevent slow subscribers from stalling each other.

The hub goroutine terminates when ctx is cancelled or src closes. All subscriber channels are then closed so downstream consumers drain cleanly.

func Apply ¶

func Apply[In, Out any](
	ctx context.Context,
	src Stream[In],
	fn *forge.Function[In, Out],
	opts ApplyOptions,
) Stream[Out]

Apply applies fn to every value in src using forge.Function.ApplyContext.

All forge validation — input codec Refine, optional WithRefinement, compute function, output codec Refine — runs per item. Successful outputs go to Stream.Values. Validation or compute failures are wrapped in StreamApplyError and sent to Stream.Errors. The stream continues after each error.

stats.PipelineObserver.RecordApply fires inside forge for every item. If opts.Observer also implements stats.StreamObserver, RecordStreamItem fires for every item with the forge function name, success flag, and duration.

The forge function's own observer (set via forge.Function.Register) fires independently — both observers can be active simultaneously.

package main

import (
	"context"
	"fmt"

	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/forge"

	stream "github.com/DaniDeer/go-codex/stream"
)

func main() {
	ctx := context.Background()
	double := forge.NewFunction("double", "1.0.0",
		codex.Float64().WithTitle("input"),
		codex.Float64().WithTitle("doubled"),
		func(v float64) (float64, error) { return v * 2, nil },
	)
	ch := make(chan float64, 3)
	ch <- 1
	ch <- 2
	ch <- 3
	close(ch)

	s := stream.Apply(ctx, stream.From(ctx, ch), double, stream.ApplyOptions{})
	vals, _ := stream.Collect(ctx, s)
	for _, v := range vals {
		fmt.Printf("%.0f\n", v)
	}
}

Output:
2
4
6

func Buffer ¶

func Buffer[T any](ctx context.Context, src Stream[T], n int, maxWait time.Duration) Stream[[]T]

Buffer collects up to n value items (or until maxWait elapses since the last emission) and emits them as a batch []T. Errors from src are forwarded to the output Stream.Errors immediately, without buffering.

Use Buffer to integrate item-at-a-time streams with forge.Map (which takes []T):

batchStream := stream.Buffer(ctx, sensorStream, 10, 500*time.Millisecond)
batchOEE    := stream.Apply(ctx, batchStream, batchOEECalc, opts)

func CombineLatest2 ¶

func CombineLatest2[A, B, Out any](
	ctx context.Context,
	a Stream[A],
	b Stream[B],
	combine func(A, B) Out,
) Stream[Out]

CombineLatest2 merges the latest value from two independent streams using combine. A new combined value is emitted whenever either source emits a new value. The output stream blocks until both sources have emitted at least one value. Errors from either source are forwarded to the output Stream.Errors.

Use CombineLatest2 to feed a forge.Function that takes a 2-field input struct, where the two fields arrive on independent streams:

oeeInputs := stream.CombineLatest2(ctx, availStream, perfStream,
    func(a Availability, p Performance) OEEIn { return OEEIn{a, p} })
oeeStream := stream.Apply(ctx, oeeInputs, oeeCalcFn, opts)

func CombineLatest3 ¶

func CombineLatest3[A, B, C, Out any](
	ctx context.Context,
	a Stream[A],
	b Stream[B],
	c Stream[C],
	combine func(A, B, C) Out,
) Stream[Out]

CombineLatest3 merges the latest value from three independent streams using combine. A new combined value is emitted whenever any source emits a new value, after all three sources have emitted at least one value. Errors from any source are forwarded to the output Stream.Errors.

Use CombineLatest3 to feed a forge function with a 3-field input struct where each field arrives on a separate stream (e.g. OEE = Availability × Performance × Quality):

oeeInputs := stream.CombineLatest3(ctx, availStream, perfStream, qualStream,
    func(a Availability, p Performance, q Quality) OEEIn {
        return OEEIn{Availability: a, Performance: p, Quality: q}
    })
oeeStream := stream.Apply(ctx, oeeInputs, oeeCalcFn, opts)

func CombineLatest4 ¶

func CombineLatest4[A, B, C, D, Out any](
	ctx context.Context,
	a Stream[A],
	b Stream[B],
	c Stream[C],
	d Stream[D],
	combine func(A, B, C, D) Out,
) Stream[Out]

CombineLatest4 merges the latest value from four independent streams. Same semantics as CombineLatest3 extended to four sources.

func Debounce ¶

func Debounce[T any](ctx context.Context, src Stream[T], d time.Duration) Stream[T]

Debounce emits a value only when src.Values is silent for at least d. Intermediate values during the silence window are dropped; only the last value before the silence elapses is emitted. Error items are forwarded to Stream.Errors immediately.

Use Debounce when only the final value of a burst matters — for example, sensor readings that settle after a transient spike.

func Filter ¶

func Filter[T any](ctx context.Context, src Stream[T], pred func(T) bool) Stream[T]

Filter keeps value items where pred returns true. Value items for which pred returns false are dropped silently. Error items are forwarded to the output Stream.Errors unchanged.

func FlatMapSlice ¶

func FlatMapSlice[In, Out any](ctx context.Context, src Stream[In], fn func(In) []Out) Stream[Out]

FlatMapSlice maps each value item to a []Out slice and emits each element of the slice individually to the output Stream.Values. An empty slice from fn produces no output items (filter-like behaviour for that item). Errors from src pass through to Stream.Errors unchanged.

Use FlatMapSlice when one incoming item should expand into multiple outgoing items — for example, one batch record expanding into N individual readings, or one sensor event triggering multiple derived measurements.

Unlike a hypothetical FlatMap[In, Stream[Out]], FlatMapSlice requires no goroutine pool: fn is called synchronously per item.

func From ¶

func From[T any](ctx context.Context, src <-chan T) Stream[T]

From wraps a typed channel as a Stream. Each value received from src becomes a value item. When src is closed or ctx is cancelled, both Stream channels are closed.

The returned Stream.Errors channel is never written — it closes when the stream terminates. This is intentional: From is a type-safe source with no error path. Use FromCodec when decode failures must be captured.

intCh := make(chan int, 3)
intCh <- 1; intCh <- 2; intCh <- 3; close(intCh)
s := stream.From(ctx, intCh)
stream.Drain(ctx, s, func(_ context.Context, v int) error {
    fmt.Println(v)
    return nil
}, nil, stream.DrainOptions{})

package main

import (
	"context"
	"fmt"

	stream "github.com/DaniDeer/go-codex/stream"
)

func main() {
	ctx := context.Background()
	ch := make(chan int, 3)
	ch <- 10
	ch <- 20
	ch <- 30
	close(ch)

	s := stream.From(ctx, ch)
	vals, _ := stream.Collect(ctx, s)
	for _, v := range vals {
		fmt.Println(v)
	}
}

Output:
10
20
30

func FromCodec ¶

func FromCodec[T any](ctx context.Context, src <-chan []byte, fmt format.Format[T], opts SourceOptions) Stream[T]

FromCodec decodes raw []byte payloads from src using the given format. Successful decodes go to Stream.Values. Decode or validation failures go to Stream.Errors as StreamDecodeError.

Pass any format.Format value — JSON, YAML, TOML, or a custom format:

sensors := stream.FromCodec(ctx, rawCh, format.JSON(sensorCodec),
    stream.SourceOptions{Name: "mqtt/sensors/+", Observer: obs})

sensors := stream.FromCodec(ctx, rawCh, format.YAML(sensorCodec),
    stream.SourceOptions{Name: "mqtt/sensors/+", Observer: obs})

Use with MQTT or ZeroMQ SubscribeHandlers that write raw payloads to a channel:

rawCh := make(chan []byte, 64)
mqttClient.Subscribe("sensors/+/data", 1,
    adaptermqtt.SubscribeHandler(ctx, handle, func(_ context.Context, raw []byte) error {
        select { case rawCh <- raw: default: }
        return nil
    }, adaptermqtt.SubscribeOptions{}))

sensors := stream.FromCodec(ctx, rawCh, format.JSON(sensorCodec),
    stream.SourceOptions{Name: "mqtt/sensors/+", Observer: obs})

package main

import (
	"context"
	"fmt"

	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/format"

	stream "github.com/DaniDeer/go-codex/stream"
	"github.com/DaniDeer/go-codex/validate"
)

type reading struct {
	Sensor string
	Value  float64
}

var readingCodec = codex.Struct(
	codex.RequiredField("sensor",
		codex.String().Refine(validate.NonEmptyString),
		func(r reading) string { return r.Sensor },
		func(r *reading, v string) { r.Sensor = v }),
	codex.RequiredField("value",
		codex.Float64(),
		func(r reading) float64 { return r.Value },
		func(r *reading, v float64) { r.Value = v }),
)

func main() {
	ctx := context.Background()
	rawCh := make(chan []byte, 2)
	rawCh <- []byte(`{"sensor":"s1","value":23.5}`)
	rawCh <- []byte(`{"sensor":"s2","value":87.3}`)
	close(rawCh)

	s := stream.FromCodec(ctx, rawCh, format.JSON(readingCodec),
		stream.SourceOptions{Name: "example"})
	vals, errs := stream.Collect(ctx, s)
	fmt.Println(len(vals), "values,", len(errs), "errors")
}

Output:
2 values, 0 errors

func MapErr ¶

func MapErr[T any](ctx context.Context, src Stream[T], fn func(error) (T, bool, error)) Stream[T]

MapErr transforms errors in Stream.Errors, enabling error recovery or reclassification.

fn receives each error and returns one of:

• (value, true, nil) — recover: emit value to Stream.Values
• (zero, false, err) — reclassify: emit new error to Stream.Errors
• (zero, false, nil) — silence: drop the error entirely

Value items pass through to the output Stream.Values unchanged.

Use MapErr for dead-lettering, retry-after-transformation, or silencing expected transient errors:

recovered := stream.MapErr(ctx, src, func(err error) (T, bool, error) {
    var sde stream.StreamDecodeError
    if errors.As(err, &sde) && isTransient(sde.Err) {
        return zero, false, nil // silence transient decode errors
    }
    return zero, false, err // re-emit all other errors
})

func Merge ¶

func Merge[T any](ctx context.Context, srcs ...Stream[T]) Stream[T]

Merge combines multiple streams into one. Items and errors from all source streams are forwarded as they arrive. The output stream terminates when all source streams have terminated.

Use Merge to combine readings from multiple sensors or topics into a single processing pipeline.

func Retry ¶

func Retry[T any](ctx context.Context, src Stream[T], retry func(error) (T, bool, error)) Stream[T]

Retry transforms error items, enabling recovery or reclassification with caller-controlled retry logic. Successful value items pass through unchanged.

retry receives each error and returns one of:

• (value, true, nil) — recover: emit value to Stream.Values
• (zero, false, err) — reclassify: emit new error to Stream.Errors
• (zero, false, nil) — silence: drop the error entirely

The caller's retry function controls timing and backoff. For exponential backoff:

retried := stream.Retry(ctx, src, func(err error) (T, bool, error) {
    var sde stream.StreamDecodeError
    if errors.As(err, &sde) {
        time.Sleep(100 * time.Millisecond) // simple backoff
        return zero, false, err            // reclassify for a retry queue
    }
    return zero, false, nil // silence unrecoverable errors
})

Retry is a specialisation of MapErr for the common case where the retry function needs a concise name. Callers who need the full (value,isValue,err) tuple directly should use MapErr.

func Single ¶

func Single[T any](ctx context.Context, v T) Stream[T]

Single wraps a single value as a Stream that emits v once, then terminates. The Stream.Errors channel is never written.

Use Single to start a per-request pipeline inside a [PipelineHandlerFunc] or [AsPipelineFunc], or any time you need a bounded one-shot stream source:

s := stream.Single(ctx, req)
out := stream.Apply(ctx, s, computeFn, stream.ApplyOptions{Observer: obs})
out = stream.Tap(ctx, out, func(v Out) { slog.Info("computed", "v", v) })

package main

import (
	"context"
	"fmt"

	stream "github.com/DaniDeer/go-codex/stream"
)

func main() {
	ctx := context.Background()

	s := stream.Single(ctx, 7)
	vals, _ := stream.Collect(ctx, s)
	fmt.Println(vals[0])
}

Output:
7

func SlidingWindow ¶

func SlidingWindow[T any](ctx context.Context, src Stream[T], size, step int) Stream[[]T]

SlidingWindow emits a []T slice every step items, containing the last size items. The window slides forward by step items on each emission. When step == size the windows are non-overlapping (tumbling). Requires step > 0 and size >= step.

Errors from src are forwarded to Stream.Errors immediately without affecting the sliding window position.

func Tap ¶

func Tap[T any](ctx context.Context, src Stream[T], onValue func(T)) Stream[T]

Tap inserts a domain event observer on the value channel without transforming items. onValue is called for each successful value; the value is then forwarded unchanged. Error items are forwarded to the output Stream.Errors unchanged.

Use Tap for domain-level observation — auditing, triggering side effects, logging application-level events — independently from infrastructure metrics:

oeeStream = stream.Tap(ctx, oeeStream, func(oee OEE) {
    slog.Info("OEE computed", "value", float64(oee))
    businessDashboard.Publish(oee)
})

package main

import (
	"context"
	"fmt"

	stream "github.com/DaniDeer/go-codex/stream"
)

func main() {
	ctx := context.Background()
	ch := make(chan int, 3)
	ch <- 10
	ch <- 20
	ch <- 30
	close(ch)

	var observed []int
	s := stream.Tap(ctx, stream.From(ctx, ch), func(v int) {
		observed = append(observed, v)
	})
	stream.Collect(ctx, s)
	fmt.Println(len(observed), "items observed")
}

Output:
3 items observed

func Throttle ¶

func Throttle[T any](ctx context.Context, src Stream[T], interval time.Duration) Stream[T]

Throttle emits at most one value per interval, dropping intermediates. The first value in an interval is emitted; subsequent values arriving before the interval elapses are dropped. Error items are forwarded to Stream.Errors immediately.

Use Throttle to rate-limit high-frequency sources while ensuring at least one value is emitted per interval.

func Window ¶

func Window[T any](ctx context.Context, src Stream[T], duration time.Duration) Stream[[]T]

Window emits all values collected during each fixed-duration time window as a []T slice. An empty slice is emitted when no items arrived during a window. Errors from src are forwarded immediately.

Unlike Buffer, Window always emits at fixed calendar-aligned intervals using time.NewTicker — the emission clock never resets when items arrive. This gives consistent time boundaries (e.g. "all readings in the past 1 minute, every minute") suitable for time-series analytics.

func Zip ¶

func Zip[A, B, Out any](
	ctx context.Context,
	a Stream[A],
	b Stream[B],
	combine func(A, B) Out,
) Stream[Out]

Zip pairs items from two streams by position: (a[0],b[0]), (a[1],b[1]), ... A combined item is emitted when both sources have emitted their n-th value. If one source emits faster, the faster source's items are buffered internally. Errors from either source are forwarded to the output Stream.Errors immediately.

Unlike CombineLatest2, which emits on every update using the latest values, Zip waits for matched pairs in order.

func NewTopology ¶

func NewTopology(title, version string) *Topology

NewTopology returns a new Topology with the given title and version.

package main

import (
	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/forge"

	stream "github.com/DaniDeer/go-codex/stream"
)

var topoFn = forge.NewFunction("oeeCalc", "1.0.0",
	codex.Float64().WithTitle("oee"),
	codex.Float64().WithTitle("grade"),
	func(v float64) (float64, error) { return v * 100, nil },
)

func main() {
	topo := stream.NewTopology("Sensor Pipeline", "1.0.0").
		WithDescription("Real-time sensor processing pipeline.").
		WithSource("mqtt/sensors/+/data", "Raw sensor readings from MQTT").
		WithFilter("value > 0").
		WithTap("dashboard observer").
		WithSink("mqtt/alerts", "OEE alert publisher")

	stream.WithApply(topo, topoFn)

	spec := topo.Spec()
	_ = spec // pass spec to render/stream.Render to get YAML
}

Output:

func WithApply ¶

func WithApply[In, Out any](t *Topology, fn *forge.Function[In, Out]) *Topology

WithApply records an apply step from a forge.Function. The function's name, version, hash, and description are captured from its forge.FunctionSpec.