Reactive Stream Pipelines — stream¶

See also: stream on pkg.go.dev · Stream Guide · Ports Guide · Ports Feature · Observer Pattern · Forge Pipelines

Runnable demos: - examples/stream-pipeline — all operators showcased: From, Apply, Tap, Filter, CombineLatest2, Tee, Merge, FlatMapSlice, Debounce, Throttle, Buffer, Window, MapErr, Switch, GroupBy, Topology + YAML render - examples/stream-oee — forge + stream integration: governed OEE from machine events (Window → governed forge chain → alert); governance YAML with SHA-256 hashes per function - examples/sensor-service — flagship port showcase: one coherent use case (MQTT ingest → SQL persist → threshold alert → REST time series → REST-triggered file export) where every IO hop is a port — mqtt.SubscribeAdapter/PublishAdapter, sql.QueryEachAdapter, file.DrainWriteFileAdapter, nethttp.PipelineAdapter, nethttp.HandlerLatest — with a single shared observer

The stream package provides a declarative reactive pipeline over typed Go channels, bridging push-based transport adapters (MQTT, ZeroMQ) with governed forge.Function computations.

forge vs stream — complementary, not competing¶

forge/ and stream/ solve different problems:

They compose: stream.Apply(ctx, mqttStream, forgeFunction, opts) — the forge function's validated computation runs per-item inside the reactive pipeline.

stream/ is a top-level package (not an adapter) because it has no external library dependency — only go-codex packages and Go stdlib.

Stream[T] — explicit error channels¶

type Stream[T any] struct {
    Values <-chan T     // successful items
    Errors <-chan error // per-item errors — stream continues after each
}

Values and errors travel on separate channels — idiomatic Go, no interface{} boxing, no type switches. A failing item (bad sensor payload, forge validation failure) does not terminate the pipeline: it goes to Errors and the pipeline continues.

Consumers MUST drain both channels concurrently. Use Drain as the safe default sink — it handles both in a single select loop.

Operators¶

Sources¶

Transforms¶

Fan-in / fan-out¶

More than 4 sources: nest CombineLatest calls (see the stream guide).

Routing¶

Time operators¶

Sinks¶

Topology (documentation)¶

Two observer kinds¶

Infrastructure metrics — stats.StreamObserver¶

type StreamObserver interface {
    RecordStreamItem(function string, success bool, dur time.Duration)
}

Type-asserted from ApplyOptions.Observer. Existing Observer implementations need not change. stats.NoopObserver, stats.LoggingObserver, and stats.NewFanout all implement it.

Domain event observation — Tap¶

Tap is a first-class operator for observing typed business values flowing through the pipeline — independent from infrastructure metrics:

oeeStream = stream.Tap(ctx, oeeStream, func(oee OEE) {
    slog.Info("OEE computed", "value", float64(oee))
    dashboard.Publish(oee)   // domain event — not a counter
})

Structured errors (all implement slog.LogValuer)¶

Both implement Error(), Unwrap(), and LogValue(). Use errors.As to reach the inner forge error from a StreamApplyError:

var sae stream.StreamApplyError
if errors.As(err, &sae) {
    slog.Warn("apply failed", "error", sae) // structured: {function, err}
    var ie forge.InputError
    if errors.As(err, &ie) {
        // per-field validation detail
    }
}

Design rationale — why forge and stream are separate packages¶

forge/ and stream/ were deliberately kept separate after an architectural evaluation during development. The key arguments:

Why not merge: 1. Different execution models — a unified type would lose either batch simplicity or reactive capability 2. Governance belongs on forge functions (signed computations), not on stream operators (Debounce, Throttle) 3. Same names, different semantics — merging would require awkward disambiguation 4. Dependency direction is one-way and correct: stream imports forge; the reverse would create a circular dependency

The correct conceptual model:

codex/      Layer 1 — validated domain types
forge/      Layer 3 — governed synchronous computation + KPI spec
stream/     Layer 4 — reactive execution of forge functions over event streams
adapters/   Transport bridges (MQTT, ZeroMQ) supply source channels to stream/

• forge/ = "what the computation is" (declarative, governed, signed)
• stream/ = "how computation runs continuously over time" (reactive, async)

They compose: stream.Apply(ctx, mqttStream, forgeFunction, opts) — the forge function's governed computation runs per-item inside the reactive pipeline.

Connecting streams to transports — the ports package¶

Rather than importing a transport package directly into pipeline code, go-codex wires streams to the outside world through ports — a protocol-agnostic binding layer. A pipeline declares a typed port; the transport is bound separately (typically in main.go), so swapping MQTT for HTTP never touches domain logic.

Full patterns and adapter catalogue: Ports Guide · Ports Feature

stream.Single[T] — one-shot source¶

s := stream.Single(ctx, req)    // emits req once, closes

Used as the entry point for per-request pipelines inside a ports.ToolPort pipeline function, or any AsPipelineFunc-wrapped handler.

The four port types, in one line each¶

// domain/pipeline.go — no adapter imports
var SensorReadings = ports.NewSourcePort[SensorReading]("sensor-readings", sensorCodec, ports.PortOptions{})
var OEEResults = ports.NewSinkPort[OEE]("oee-results", oeeCodec, ports.PortOptions{})

func StartPipeline(ctx context.Context) {
    sensors := SensorReadings.Stream(ctx)
    oeeStream := stream.Apply(ctx, sensors, oeeCalcFn, stream.ApplyOptions{})
    go OEEResults.Feed(ctx, oeeStream)
}

// main.go — bind concrete transports here
domain.SensorReadings.Bind(ctx, mqtt5.SubscribeAdapter(client, router, sensorHandle, 0, format.JSON(sensorCodec), opts))
domain.OEEResults.Bind(ctx, mqtt5.PublishAdapter(client, alertHandle, format.JSON(oeeCodec), publishOpts))
domain.OEEResults.Bind(ctx, nethttp.SSEAdapter(mux, sseHandle, nethttp.SSEAdapterOptions{})) // fan-out

Request/response pipelines with ToolPort¶

ToolPort[In,Out] sets the pipeline function once and can bind it to multiple transports simultaneously — the same domain logic serves MCP, HTTP, and ZeroMQ:

var OEETool = ports.NewToolPort[OEEIn, OEEResult]("oee-calc", oeeInCodec, oeeCodec, ports.PortOptions{})

func init() {
    OEETool.SetPipeline(func(ctx context.Context, req OEEIn) stream.Stream[OEEResult] {
        s := stream.Single(ctx, req)
        s = stream.Apply(ctx, s, validateFn, stream.ApplyOptions{})
        s = stream.Tap(ctx, s, func(v ValidatedReq) { slog.Info("validated", "id", v.ID) })
        return stream.Apply(ctx, s, oeeCalcFn, stream.ApplyOptions{})
    })
}

domain.OEETool.Bind(ctx, mcpgo.ToolPipelineAdapter(mcpServer, mcpToolHandle, mcpgo.Options{}))
domain.OEETool.Bind(ctx, nethttp.PipelineAdapter(mux, httpHandle, nethttp.PipelineAdapterOptions{}))
domain.OEETool.Bind(ctx, zeromq.ServeAdapter(repSock, zmqHandle, zeromq.ServeOptions{}))

New error types: ports.PortBindError{Port,Adapter,Err}, ports.PortNoAdapterError{Port}, ports.PortNoPipelineError{Port} — all implement slog.LogValuer. Transport-level errors (NoLatestValueError, PipelineFullError, SSEWriteError, …) are unchanged and still surface through the same adapters.

I/O role taxonomy — source, intermediate, sink¶

Every I/O operation in a stream pipeline has a role. Recognising the role guides which API to use:

The fundamental rule: forge functions are pure — zero I/O¶

stream.Apply accepts *forge.Function[In, Out]. These functions are governed domain computations — their only job is to transform typed inputs into typed outputs. They must have no I/O.

// ✅ Correct: forge function is pure
oeeCalcFn := forge.NewFunction("oeeCalc", "1.0.0", oeeInCodec, oeeCodec,
    func(in OEEIn) (OEE, error) {
        // Pure calculation — no file reads, no HTTP calls, no database queries
        return OEE(float64(in.Availability) * float64(in.Performance) * float64(in.Quality)), nil
    })

// ❌ Wrong: I/O inside a forge function violates the design principle
// func(in InputData) (Out, error) {
//     cfg, _ := configFile.Read(...)  // ← I/O inside forge
//     return combine(in, cfg), nil
// }

I/O that a forge function needs (e.g. a config lookup) must arrive as typed input — declared in the input codec and flowing from the stream layer:

// Config is an explicit input — loaded by the stream layer, not by the function:
type EnrichInput struct {
    Data   InputData
    Config ThresholdConfig   // explicit input — arrives via CombineLatest2
}

enrichFn := forge.NewFunction("enrich", "1.0.0", enrichInputCodec, outputCodec,
    func(in EnrichInput) (Out, error) {
        return combine(in.Data, in.Config), nil // pure: all inputs explicit
    })

// I/O stays in the stream layer:
configs := stream.Single(ctx, preloadedConfig)  // or a SourcePort bound to file.WatchAdapter for dynamic reload
combined := stream.CombineLatest2(ctx, dataStream, configs,
    func(d InputData, c ThresholdConfig) EnrichInput { return EnrichInput{d, c} })
enriched := stream.Apply(ctx, combined, enrichFn, stream.ApplyOptions{})

When to use each I/O container¶

Use ports.IOPort[Req,Resp] when the step's whole purpose is an I/O call — send a request, receive a response, forward the response. Bind exactly one adapter; swapping the adapter (HTTP → SQL → file) never touches the pipeline:

var Enrichment = ports.NewIOPort[Raw, Enriched]("enrichment", rawCodec, enrichedCodec, ports.PortOptions{})

// main.go — pick one:
domain.Enrichment.Bind(ctx, zeromq.CallAdapter(sock, enrichHandle, zeromq.CallStreamOptions{}))
// domain.Enrichment.Bind(ctx, mqtt5.CallAdapter(client, router, enrichHandle, mqtt5.CallOptions{}))
// domain.Enrichment.Bind(ctx, nethttp.CallAdapter(httpClient, baseURL, enrichHandle, nethttp.CallStreamOptions{}))
// domain.Enrichment.Bind(ctx, sql.QueryEachAdapter(enrichedCodec, queryFn, sql.QueryEachStreamOptions{}))

enriched := domain.Enrichment.Connect(ctx, rawStream)

I/O is explicit in the pipeline declaration. Errors are transport-typed, not wrapped in forge.ApplyError.

Use stream.Tap for side-effect I/O that produces no value for the pipeline — publishing, writing, logging:

results = stream.Tap(ctx, results, func(r OEEResult) {
    mqtt.Publish(ctx, client, alertHandle, 0, false, r, nil, mqtt.PublishOptions{}) // fire-and-forget
    resultFile.Write(nil, r, ports.FileOptions{})                                 // whole-file write
})

Use stream.FlatMapSlice for ad-hoc transforms that have I/O and are not governed (no forge hash, no schema validation — use when governance is not required):

// I/O in stream.FlatMapSlice is in the stream layer, not forge:
enriched := stream.FlatMapSlice(ctx, ids, func(id string) []Data {
    v, err := someFile.Read(map[string]string{"id": id}, opts)
    if err != nil { return nil } // errors silently dropped — use ports.IOPort for typed errors
    return []Data{v}
})

All transports have an intermediate transform operator¶

Every transport with a request/response protocol provides an IOAdapter for ports.IOPort — HTTP, ZeroMQ, MQTT5, SQL, and File all support per-item enrichment:

zeromq.CallAdapter      ✅  — send each item as REQ, emit each REP response
mqtt5.CallAdapter       ✅  — send each item as MQTT5 request-reply, emit responses
nethttp.CallAdapter     ✅  — full HTTP codec machinery (path vars, query/cookie/header
                              params, security, structured errors) per item
sql.QueryEachAdapter    ✅  — parameterized SQL query per item (1:N rows)
file.ReadEachAdapter    ✅  — per-item file read with path template vars

See the Ports Guide for the complete adapter catalogue.