Reactive Streams

GoAkt Streams is a reactive, demand-driven data-processing library built directly on top of the GoAkt actor model. Every pipeline stage runs inside a GoAkt actor, which means it inherits supervision, lifecycle management, and location transparency for free. Pipelines are lazy: nothing executes until RunnableGraph.Run is called against a live ActorSystem.

Why

Raw actor messaging requires you to handle buffer overflow, message loss, and backpressure by hand: every actor that consumes faster or slower than its peers becomes a coordination problem you have to solve case-by-case. Streams handle all of this automatically, giving you a safe and composable programming model on top of actors so you can focus on what flows through the pipeline rather than how to keep it from breaking under load.

Overview

Correctness: Exactly-once delivery semantics within a local pipeline.
Backpressure: Credit-based demand pull prevents unbounded buffering.
Composability: Stages are plain Go values; pipelines are assembled declaratively.
Actor-native: Every stage is an actor; supervision, scheduling, and routing apply naturally.
Type-safety: Source[T], Flow[In, Out], Sink[T] carry type parameters through the pipeline.

Core abstractions

Source[T]: Origin of stream elements; produces values of type T.
Flow[In, Out]: Transformation stage; consumes In, emits Out.
Sink[T]: Terminal consumer; processes elements and drives backpressure.
SubFlow[K, T]: “Stream of streams”: partition by key, transform per-substream, merge back.
RunnableGraph: Fully assembled pipeline; a value type that can be Run multiple times.
StreamHandle: Live handle returned by Run; controls lifecycle and exposes metrics.

A minimal pipeline looks like this:

Core abstractions

import "github.com/tochemey/goakt/v4/stream"

collector, sink := stream.Collect[int]()

handle, err := stream.From(stream.Of(1, 2, 3, 4, 5)).
    Via(stream.Filter[int](func(n int) bool { return n%2 == 0 })).
    Via(stream.Map(func(n int) int { return n * 10 })).
    To(sink).
    Run(ctx, sys)
if err != nil {
    return err
}

<-handle.Done()
fmt.Println(collector.Items()) // [20 40]

Quick start

Linear pipeline

Use the fluent From / Via / To builder for straight-line pipelines:

Linear pipeline

foldResult, sink := stream.Fold(0, func(acc, n int) int { return acc + n })

handle, err := stream.From(stream.Range(1, 101)).  // 1..100
    Via(stream.Filter[int64](func(n int64) bool { return n%2 == 0 })).
    Via(stream.Map(func(n int64) int { return int(n) })).
    To(sink).
    Run(ctx, sys)
if err != nil {
    return err
}

<-handle.Done()
fmt.Println(foldResult.Value()) // 2550

Type-changing flows

When a Flow changes the element type, use the package-level Via free function or ViaLinear:

Type-changing flows

// Via free function
strSrc := stream.Via(
    stream.From(stream.Of("1", "2", "3")).Source(),
    stream.Map(func(s string) int { n, _ := strconv.Atoi(s); return n }),
)

// ViaLinear on a LinearGraph
g := stream.ViaLinear(
    stream.From(stream.Of("1", "2", "3")),
    stream.Map(func(s string) int { n, _ := strconv.Atoi(s); return n }),
)

Sources

Sources are the origin of stream data. All sources apply backpressure: they produce only as many elements as downstream demands.

Finite sources

// Fixed values
stream.Of(1, 2, 3)

// Integer range [start, end)
stream.Range(0, 1000)

// Seed + step function (Fibonacci)
stream.Unfold([]int{0, 1}, func(s []int) ([]int, int, bool) {
    return []int{s[1], s[0] + s[1]}, s[0], true
})

Channel source

Bridges an external Go channel into the pipeline. Backpressure naturally limits how fast the goroutine sends:

Channel source

ch := make(chan string, 100)
go feedChannel(ch) // your producer

handle, err := stream.From(stream.FromChannel(ch)).
    To(stream.ForEach(func(s string) { log.Println(s) })).
    Run(ctx, sys)

Actor source

Pulls elements from a GoAkt actor using the PullRequest / PullResponse[T] protocol. Useful for integrating existing actor-based producers:

Actor source

// Your actor must handle *stream.PullRequest and reply with *stream.PullResponse[T].
handle, err := stream.From(stream.FromActor[Order](orderActorPID)).
    To(stream.ForEach(func(o Order) { process(o) })).
    Run(ctx, sys)

Tick source

Emits the current time on a fixed interval. Runs indefinitely until Stop or Abort is called:

Tick source

handle, err := stream.From(stream.Tick(time.Second)).
    Via(stream.Map(func(t time.Time) string { return t.Format(time.RFC3339) })).
    To(stream.ForEach(func(s string) { log.Println(s) })).
    Run(ctx, sys)

time.Sleep(10 * time.Second)
handle.Stop(ctx)

Network source

Reads []byte frames from a net.Conn. Each Read call produces one element; demand controls how many reads are batched per upstream request:

Network source

conn, _ := net.Dial("tcp", "host:port")
handle, err := stream.From(stream.FromConn(conn, 4096)).
    Via(stream.Map(func(b []byte) string { return string(b) })).
    To(stream.ForEach(func(s string) { log.Println(s) })).
    Run(ctx, sys)

Reference

Of[T]: Finite source from a fixed set of values.
Range: Integer range [start, end).
Unfold[S, T]: Generates values from a seed with a stateful step function.
FromChannel[T]: Reads from a Go channel; completes when the channel closes.
FromActor[T]: Pulls from a GoAkt actor via PullRequest / PullResponse[T].
Tick: Emits time.Time on a fixed interval; runs until cancelled.
Merge[T]: Fans N sources into one; completes when all inputs complete.
Concat[T]: Consumes sub-sources sequentially; completes when the last completes.
Combine[T, U, V]: Zips two sources pairwise via fn (mixed input types).
Zip[T]: Zips N same-typed sources into []T tuples.
ZipWith[T, V]: N-input zip via user combine fn.
MergePreferred[T]: Like Merge but drains slot idx first when it has data.
MergePrioritized[T]: Weighted random slot selection over slots that have data.
MergeLatest[T]: Emits []T snapshots of the latest value on every input.
MergeSequence[T]: Emits in ascending seq order across N inputs.
Broadcast[T]: Fans one source out to N independent branches.
Balance[T]: Distributes one source across N branches with round-robin backpressure.
Partition[T]: Routes each element to one of N branches via predicate fn.
Unzip[T, A, B]: Splits a source into two sources of different element types.
FromConn: Reads []byte frames from a net.Conn.

Flows

Flows are lazy transformation stages. Each flow operator returns a new Flow value; multiple flows can be chained without materializing anything.

Mapping and filtering

// Map: type-changing, no error path
stream.Map(func(n int) string { return strconv.Itoa(n) })

// TryMap: transformation that may return an error
stream.TryMap(func(s string) (int, error) { return strconv.Atoi(s) })

// Filter: keeps elements where predicate returns true
stream.Filter(func(n int) bool { return n > 0 })

// FlatMap: expands each element into a slice
stream.FlatMap(func(s string) []string { return strings.Split(s, ",") })

// Flatten: unwraps []T elements into individual elements
stream.Flatten[string]()

Streaming flat-map

When the per-element expansion is itself a stream (e.g. paginated API calls, per-row database queries), use the streaming flat-map operators instead of FlatMap. Each input element produces a Source[Out] whose elements are flattened into the main pipeline:

Streaming flat-map

// FlatMapConcat: drain each sub-source fully before starting the next.
// Order is preserved end-to-end.
stream.FlatMapConcat(func(userID int) stream.Source[Order] {
    return fetchUserOrders(userID) // Source[Order]
})

// FlatMapMerge: up to N sub-sources active concurrently; outputs interleaved.
// Order across sub-sources is non-deterministic; order within each preserved.
stream.FlatMapMerge(8, func(userID int) stream.Source[Order] {
    return fetchUserOrders(userID)
})

Backpressure is bounded by the breadth: the FlatMap stage requests at most breadth elements from upstream, and the next batch is requested only as sub-sources complete: so a slow inner source naturally throttles the outer pipeline. Sub-pipeline handles are tracked and aborted on cancellation or failure to avoid leaking goroutines.

Windowing and rate control

// Batch: groups into slices of at most n, flushing early after maxWait
stream.Batch[Event](100, 500*time.Millisecond)

// Throttle: limits throughput to n elements per duration
stream.Throttle[Request](100, time.Second)

// Buffer: asynchronous decoupling buffer; when full it backpressures upstream
stream.Buffer[int](256, stream.DropTail)

Stateful flows

// Deduplicate: suppresses consecutive duplicates (T must be comparable)
stream.Deduplicate[string]()

// Scan: running accumulation; emits each intermediate state
stream.Scan(0, func(acc, n int) int { return acc + n })

Parallel processing

ParallelMap applies a function concurrently with up to n goroutines. Use OrderedParallelMap when output order must match input order:

Parallel processing

// Unordered: results arrive as goroutines complete
stream.ParallelMap(8, func(req Request) Response { return callAPI(req) })

// Ordered: min-heap resequences results to preserve input order
stream.OrderedParallelMap(8, func(req Request) Response { return callAPI(req) })

Reference

Map[In, Out]: Type-changing transformation; no error path.
TryMap[In, Out]: Transformation with error; ErrorStrategy controls failure handling.
Filter[T]: Keeps only elements where predicate returns true.
FlatMap[In, Out]: Expands each element into a slice of outputs.
FlatMapConcat[In, Out]: Per-element streaming flat-map; sub-sources drained sequentially.
FlatMapMerge[In, Out]: Per-element streaming flat-map; up to breadth sub-sources concurrent.
Flatten[T]: Unwraps []T elements into individual elements.
Batch[T]: Groups into []T slices of at most n; flushes early after maxWait.
Buffer[T]: Asynchronous buffer of the given capacity; a full buffer applies backpressure.
Throttle[T]: Limits throughput to at most n elements per per duration.
Deduplicate[T]: Suppresses consecutive duplicate elements (T must be comparable).
Scan[In, State]: Running accumulation; emits each intermediate state.
WithContext[T]: Passes elements through unchanged. The key/value pair is reserved for future tracing and is not recorded yet.
ParallelMap[In, Out]: Applies fn concurrently with up to n goroutines; unordered output.
OrderedParallelMap[In, Out]: Like ParallelMap but preserves input order via min-heap resequencing.

Sinks

Sinks are the terminal consumers of a pipeline. They drive backpressure by signalling demand upstream; the pipeline does not produce faster than the sink can consume.

Common sinks

// ForEach: side-effect per element
stream.ForEach(func(n int) { log.Println(n) })

// Collect: accumulates all elements; read results after completion
collector, sink := stream.Collect[int]()
// ... run pipeline ...
<-handle.Done()
items := collector.Items()

// Fold: reduces to a single accumulated value
result, sink := stream.Fold(0, func(acc, n int) int { return acc + n })
// ... run pipeline ...
<-handle.Done()
total := result.Value()

// First: captures only the first element then cancels upstream
first, sink := stream.First[int]()
// ... run pipeline ...
<-handle.Done()
val := first.Value()

// Ignore: discards all elements (useful when side effects are in upstream flows)
stream.Ignore[int]()

Actor and channel sinks

// ToActor: forwards each element to a GoAkt actor via Tell
stream.ToActor[Event](handlerPID)

// ToActorNamed: resolves actor by name on each element and forwards
stream.ToActorNamed[Event](sys, "handler")

// Chan: writes to a Go channel; blocks when full (natural backpressure)
ch := make(chan int, 64)
stream.Chan(ch)

Reference

ForEach[T]: Calls fn for each element.
Collect[T]: Accumulates all elements; retrieve via Collector[T].Items().
Fold[T, U]: Reduces to a single value; retrieve via FoldResult[U].Value().
First[T]: Captures the first element then cancels upstream.
Ignore[T]: Discards all elements.
Chan[T]: Writes each element to a Go channel; applies backpressure when full.
ToActor[T]: Forwards each element to a GoAkt actor via Tell.
ToActorNamed[T]: Resolves actor by name and forwards via Tell.

Fan-out and fan-in

Broadcast

Broadcast fans a single source out to N independent branches. Every branch receives every element. Backpressure is enforced by the slowest branch: the hub pulls from upstream only when all active branches have outstanding demand.

Broadcast

branches := stream.Broadcast[Event](stream.FromChannel(eventCh), 2)

// Branch 1: write to database
handle1, _ := stream.From(branches[0]).
    To(stream.ForEach(func(e Event) { db.Insert(e) })).
    Run(ctx, sys)

// Branch 2: publish to Kafka
handle2, _ := stream.From(branches[1]).
    To(stream.ForEach(func(e Event) { kafka.Publish(e) })).
    Run(ctx, sys)

Both pipelines must be materialized. The upstream sub-pipeline starts once the last branch materializes. The branch graphs can be run again. Each run of all the branches is a generation with its own hub and its own materialization of the upstream: the k-th run of each branch is paired with the k-th run of the others. A branch that is run without its siblings waits for them; Stop or Abort on its handle ends the wait. A branch aborted before it registered leaves its siblings in the same position: they wait for a new partner, and Stop or Abort on their handles ends that wait. A branch that is stopped or aborted after it registered is released by the hub and does not hold its siblings back; when every branch is gone the hub cancels the upstream. If the upstream cannot be materialized, every branch of that generation fails with the materialization error. The same applies to Balance, Partition and fan-out nodes of the Graph DSL.

Balance

Balance distributes elements across N branches using round-robin routing with backpressure. Each element goes to exactly one branch: the next one with available demand. Use this to parallelise work across independent consumers:

Balance

workers := stream.Balance[Job](stream.FromChannel(jobCh), 4)

for i, branch := range workers {
    i := i
    stream.From(branch).
        To(stream.ForEach(func(j Job) { process(i, j) })).
        Run(ctx, sys)
}

Partition

Partition routes each element to exactly one of N branches based on a user predicate. Out-of-range or already-cancelled slot results are dropped silently. Backpressure is conservative: the hub pulls from upstream only when every active branch has outstanding demand, so the destination slot for any incoming element is guaranteed capacity.

Partition

branches := stream.Partition(stream.FromChannel(eventCh), 3, func(e Event) int {
    switch e.Kind {
    case "audit":   return 0
    case "metric":  return 1
    default:        return 2
    }
})

for i, b := range branches {
    stream.From(b).To(stream.ForEach(handlers[i])).Run(ctx, sys)
}

Unzip

Unzip splits a single source into two sources of potentially different element types. It is composed of Broadcast(src, 2) plus a Map per branch and follows the same backpressure rules as Broadcast.

Unzip

type kv struct { K string; V int }

keys, values := stream.Unzip(
    stream.FromChannel(kvCh),
    func(p kv) (string, int) { return p.K, p.V },
)

Merge

Merge fans N independent sources into a single downstream. Elements arrive in non-deterministic order; completion happens when all inputs have completed. If an input fails, the merged stream fails with that input’s error and the remaining inputs are stopped; every fan-in source (Merge, MergePreferred, MergePrioritized, MergeLatest, MergeSequence, Concat, Zip, ZipWith, Combine) behaves this way. Stopping or aborting the stream also stops its inputs. Fan-in sources respect backpressure: each input is pulled for at most one demand window (224 elements) ahead of what the downstream has consumed. Two exceptions follow from their semantics: MergeLatest keeps only the latest value of each input, so it never holds an input back, and MergeSequence reads on without bound while the next sequence number is missing:

Merge

merged := stream.Merge(
    stream.FromChannel(highPriorityCh),
    stream.FromChannel(lowPriorityCh),
)

stream.From(merged).
    To(stream.ForEach(func(e Event) { handle(e) })).
    Run(ctx, sys)

Concat

Concat consumes sub-sources sequentially: the next sub-pipeline only spawns after the current one completes, bounding in-flight cost at one sub-pipeline regardless of source count. Per-source ordering is preserved across the boundary.

Concat

all := stream.Concat(
    stream.Of(1, 2, 3),
    stream.Of(4, 5),
    stream.Of(6, 7, 8, 9),
)
// emits: 1, 2, 3, 4, 5, 6, 7, 8, 9

Combine (zip, two-input mixed-type)

Combine pairs elements from two sources with a combine function. It completes when either source is exhausted. Use it when the two inputs have different element types:

Combine (zip, two-input mixed-type)

prices := stream.FromChannel(priceCh)    // Source[Price]
orders := stream.FromChannel(orderCh)    // Source[Order]

quotes := stream.Combine(prices, orders, func(p Price, o Order) Quote {
    return Quote{Price: p, Order: o}
})

Zip / ZipWith (N-input same-type)

Zip combines N same-typed sources into a stream of []T tuples; ZipWith does the same with a user combine function. Both complete as soon as any input is exhausted.

Zip / ZipWith (N-input same-type)

// []int snapshots, one per matched group
trios := stream.Zip(
    stream.Of(1, 2, 3),
    stream.Of(10, 20, 30),
    stream.Of(100, 200, 300),
)

// User combine: sum across slots
sums := stream.ZipWith(
    func(parts []int) int { s := 0; for _, v := range parts { s += v }; return s },
    stream.Of(1, 2, 3),
    stream.Of(10, 20, 30),
    stream.Of(100, 200, 300),
)

MergePreferred

Like Merge, but always drains the priority slot first when it has data, falling back to the lowest-indexed non-empty slot otherwise. Useful when one input represents a higher-priority feed (e.g. a control channel) that should be drained ahead of background traffic. Panics on out-of-range index.

MergePreferred

src := stream.MergePreferred(
    0, // slot 0 = control
    stream.FromChannel(controlCh),
    stream.FromChannel(workCh),
)

MergePrioritized

Picks the next slot by weighted random over slots that currently have data. Slots with weight 0 are only selected when they are the only ones with data. Panics on weights / sources length mismatch.

MergePrioritized

src := stream.MergePrioritized(
    []int{3, 1}, // src1 wins ~3:1
    src1,
    src2,
)

MergeLatest

Emits []T snapshots of the latest value seen on every input. Each upstream emission queues one snapshot, but the first only fires once every input has emitted at least once. Completes when every input completes.

MergeLatest

snapshots := stream.MergeLatest(
    stream.FromChannel(temperatureCh),
    stream.FromChannel(humidityCh),
)

stream.From(snapshots).
    To(stream.ForEach(func(s []float64) {
        log.Printf("temp=%.1f humidity=%.1f", s[0], s[1])
    })).
    Run(ctx, sys)

MergeSequence

Emits in strict ascending sequence-number order across N inputs by buffering out-of-order elements in a min-heap. Inputs must collectively produce a contiguous range of sequence numbers starting from 0. If all inputs complete with the next-expected sequence still missing, the stream fails with a missing-sequence error.

MergeSequence

type item struct { Seq int64; Val string }

ordered := stream.MergeSequence(
    func(i item) int64 { return i.Seq },
    evenSeqSrc, // emits seqs 0, 2, 4, ...
    oddSeqSrc,  // emits seqs 1, 3, 5, ...
)

Substreams (SubFlow)

Three constructors produce a SubFlow, each with different routing semantics: GroupBy partitions by a key function (one substream per distinct key, materialised lazily), SplitWhen starts a new substream when a predicate is true (the triggering element becomes the FIRST of the new substream), and SplitAfter ends the current substream after a predicate-true element (the triggering element becomes the LAST of the closing substream). Per-substream state (Scan accumulators, Deduplicate’s last-seen value, anything stateful in the per-substream chain) is genuinely independent because each substream is a real GoAkt actor pipeline materialised lazily on its first element. MergeSubstreams collapses the partitioned stream back into a flat Source[T] whose elements interleave non-deterministically across substreams while preserving order within a single substream. A failure of the pipeline that feeds the splitter fails the merged stream with that error, and when the merged stream ends, by completion, failure, Stop or Abort, the feeding pipeline and every open substream are stopped with it.

Substreams (SubFlow)

// Partition events by tenantID, count events per tenant via a per-substream
// Scan, then merge all per-tenant counters into a single flat stream.
src := stream.FromChannel(eventCh) // Source[Event]

byTenant := stream.GroupBy(src, 1024, func(e Event) string { return e.TenantID })
counted  := stream.SubFlowVia(byTenant, stream.Scan(0, func(acc int, _ Event) int { return acc + 1 }))
merged   := stream.MergeSubstreams(counted)

stream.From(merged).
    To(stream.ForEach(func(n int) { log.Println(n) })).
    Run(ctx, sys)

The key function is captured at GroupBy time and applied to upstream elements regardless of how many SubFlowVia calls follow: the per-substream output type can change freely without affecting routing.

Splitting on a delimiter

SplitWhen and SplitAfter take a predicate instead of a key function and produce a SubFlow[int, T] whose synthetic substream IDs increment on each rotation. They differ in where the triggering element lands:

Splitting on a delimiter

// SplitWhen: predicate-true element starts a NEW substream.
//   input:        1, 2, 0, 3, 4, 0, 5
//   substreams:  [1, 2] [0, 3, 4] [0, 5]
src := stream.FromChannel(eventCh) // Source[int]
sf  := stream.SplitWhen(src, func(n int) bool { return n == 0 })

// SplitAfter: predicate-true element ENDS the current substream.
//   input:        1, 2, 0, 3, 4, 0, 5
//   substreams:  [1, 2, 0] [3, 4, 0] [5]
sf  := stream.SplitAfter(src, func(n int) bool { return n == 0 })

These are the natural primitives for parsing delimited streams (newline-terminated frames, control sentinels, batch markers) where each delimiter starts or ends a logical group rather than tagging a key. Substreams are sequential (at most one is actively receiving new elements at a time), so the maxSubstreams cap is generally unnecessary and 0 (unbounded) is fine.

Cardinality cap

The maxSubstreams parameter caps the number of concurrently-open substreams; pass 0 for unbounded. Exceeding the cap terminates the stream with ErrTooManySubstreams, so unbounded key cardinality is a visible failure rather than a silent memory leak:

Cardinality cap

sf := stream.GroupBy(src, 100, keyFn) // up to 100 distinct keys

Per-substream backpressure

The merged stream honors the demand of its downstream. Each substream runs at most one demand window (224 elements) ahead of what the downstream has taken, and the splitter stops taking elements from the source while its merged buffer holds a window. A slow downstream therefore holds the source back, whatever the overflow strategy, and nothing is dropped for it. Each substream also has an independent in-flight cap (default 256 elements). When a substream is slower than the source feeds it and reaches the cap, the configured OverflowStrategy decides what happens to new elements with that key. Substream feed sources acknowledge consumption to the splitter in batches (matching the flowActor watermark refill at perKeyBuffer/4), so per-key memory is bounded without per-element protocol overhead.

Per-substream backpressure

sf := stream.GroupBy(src, 0, keyFn).
    WithSubstreamBuffer(64, stream.FailSource)

BackpressureSource (default): Hold the source back until the substream has room. Nothing is lost, but the other substreams wait too, because the source is one ordered stream: one slow key stalls the rest.
DropTail / DropHead: Discard the newest element when the substream is at capacity. DropHead is treated as drop-newest, because dropping the oldest would need a queue per key.
FailSource: Terminate the stream with ErrSubstreamOverflow.

The strategy applies only while the downstream asks for elements, so a slow downstream never causes a drop or a failure. A key that carries most of the elements can still reach the cap in bursts while the downstream keeps up: the source feeds one substream faster than the substream hands its elements on. Under DropTail such a key loses some elements and under FailSource the stream can fail. Keep the default BackpressureSource to lose nothing.

Drops are reported through the existing OnDrop hook and the droppedElements metric. For more elaborate pacing per key (e.g. time-based), compose a Throttle flow inside the per-substream chain.

Per-substream error strategy

When a per-key sub-pipeline fails, SubstreamErrorStrategy controls how the splitter reacts. The default mirrors the FailFast contract of linear flows; the alternatives let one substream’s failure stay isolated.

SubstreamFailAll (default): Surface the error verbatim via StreamHandle.Err and terminate the entire stream.
SubstreamDrop: Blocklist the failing key and complete that substream silently; sibling substreams finish normally; further elements with the blocklisted key are dropped.
SubstreamRestart: Discard the failed pipeline; the next element with that key spawns a fresh substream from scratch.

Per-substream error strategy

sf := stream.GroupBy(src, 0, keyFn).
    WithErrorStrategy(stream.SubstreamDrop)

Reference

GroupBy[T, K]: Partition src by keyFn; returns SubFlow[K, T].
SplitWhen[T]: New substream begins when predicate is true; element is FIRST of new substream.
SplitAfter[T]: Current substream ends after predicate is true; element is LAST of that substream.
SubFlowVia[K, In, Out]: Append a transformation to each substream’s pipeline.
MergeSubstreams[K, T]: Collapse substreams back into a flat Source[T] (non-deterministic interleave).
SubFlow.WithSubstreamBuffer: Per-substream in-flight cap and overflow strategy.
SubFlow.WithErrorStrategy: Per-substream failure handling policy.

Pipeline DSL (Graph builder)

For non-linear topologies (fan-out branches that later merge, or multiple independent pipelines sharing a source) use the Graph DSL. Nodes are identified by string names and connected by referencing upstream names:

Pipeline DSL (Graph builder)

// The Graph DSL is type-erased: nodes operate on Source[any], Flow[any, any],
// and Sink[any]. Use these typed constructors when assembling.
g := stream.NewGraph().
    AddSource("in",     stream.FromChannel[any](eventCh)).
    AddFlow("validate", stream.Filter[any](isValid),         "in").
    AddFlow("enrich",   stream.Map[any, any](enrich),        "validate").
    AddSink("db",       stream.ForEach[any](db.Insert),      "enrich").
    AddSink("metrics",  stream.ForEach[any](metrics.Record), "enrich")

rg, err := g.Build()
if err != nil { return err }

handle, err := rg.Run(ctx, sys)

Two fan-in node kinds are available alongside AddFlow and AddSink:

MergeInto(name, from...) interleaves elements from multiple upstream nodes (non-deterministic order).
ConcatInto(name, from...) consumes upstreams sequentially, preserving per-source ordering across the boundary.

Pipeline DSL (Graph builder)

g := stream.NewGraph().
    AddSource("a",   srcA).
    AddSource("b",   srcB).
    ConcatInto("ab", "a", "b").     // a then b, in order
    AddSink("out",   sink, "ab")

Cross-node stream refs

SourceRef[T] and SinkRef[T] are wire-portable handles to a stream endpoint actor. Pass one inside any registered remote message and a different node materialises it back into an ordinary Source[T] or Sink[T]: the same fluent builder works whether the producer and consumer share an actor system or live on opposite ends of a cluster. The model mirrors Akka StreamRefs. Refs encode the producer node’s address (host, port, actor name), so cross-node resolution is a direct RemoteLookup against the producer’s remote server. There is no reliance on cluster-registry replication: the consumer can resolve and subscribe the moment the producer node is reachable, regardless of how loaded the cluster’s actor registry is.

Setup

Cross-node refs only work when both sides of the connection have the wire control protocol and the element type registered with their remoting layer. Without this registration the consumer’s bridge can’t subscribe (control messages won’t deserialize) or the producer’s elements can’t be serialised: symptoms include hanging Run calls, stream-level "no serializer found for message type ..." errors, and elements silently routed to dead-letter. Required on every node that participates:

Setup

remoteCfg := remote.NewConfig("0.0.0.0", 9000,
    stream.RemoteOptions(),                  // (1) control-plane wire types
    remote.WithSerializables(new(MyEvent)),  // (2) your element type T
)

sys, err := actor.NewActorSystem("my-app",
    actor.WithRemote(remoteCfg),             // remoting must be enabled
    actor.WithCluster(clusterCfg),           // for cluster-based discovery
)

stream.RemoteOptions() registers the small subscribe / request / complete / error / cancel control-plane wire types. Append it to your existing remote.NewConfig options on every node: producer side, consumer side, and any node that might hold or forward a ref. Forgetting it on either side is the most common cause of hangs: the bridge sends streamSubscribeWire and the endpoint never sees a recognisable message because its remoting layer can’t decode the type.
Element types must be registered with remote.WithSerializables(new(MyEvent)): same as for any other rctx.Tell to a remote PID. Refs do not wrap elements in an envelope, so missing element-type registration on the producer side fails the send, and missing it on the consumer side fails the receive.

Things that will silently break the feature otherwise:

Asymmetric registration. The producer registers MyEvent but the consumer does not (or vice versa). The end with the missing registration logs a "no serializer found" warning when the message arrives; for the consumer that means elements never reach the local sink. Always register the same element types on both ends.
Registering the value type as a non-pointer. remote.WithSerializables keys on the exact type passed in: typically *MyEvent from new(MyEvent). Sending a value-type element over the wire still works because the bridge’s wireForm helper auto-wraps non-pointer values into a *T registered key, but if you pass a struct value (instead of new(...) or an interface pointer) into WithSerializables itself, the registry stores the wrong type and lookups miss. Stick to the new(MyEvent) form.
Skipping stream.RemoteOptions() on a forwarding-only node. Any node that may receive a SourceRef[T] / SinkRef[T] inside another message (even just to inspect or relay it) needs stream.RemoteOptions() registered, because the ref struct itself is serialised through the same registry. A node missing it will fail to deserialize incoming refs.
Disabling remoting entirely. Cross-node refs require actor.WithRemote(...) on both ends. Without it ref.Source(sys) / ref.Sink(sys) materialise a bridge that can’t reach the remote endpoint and surfaces an "actor not found" resolution error after the retry budget expires.
Producer node unreachable at resolve time. Refs carry the producer node’s host:port, so resolution is a direct RemoteLookup against the producer’s remote server: there is no dependency on async cluster registry propagation. If the producer node is briefly unreachable when the consumer resolves (network blip, just-restarting node), the bridge retries with jittered exponential backoff for up to 10s. After that it surfaces a "resolve source ref" / "resolve sink ref" error rather than hanging: handle it in the receiving side just like any remote-Tell failure.

Element values cross the wire as ordinary remote messages: no extra envelope, no double encoding. Whatever is registered with remote.WithSerializables round-trips automatically, just as it would for any other rctx.Tell to a remote PID.

`SourceRef`: publish a producer

A SourceRef[T] exposes a local Source[T] to a remote consumer. The ref is a plain serialisable value: wrap it in any user-defined message type you like (registered with the same remoting layer) and ship it however your application already moves data between nodes.

SourceRef: publish a producer

// User-defined wrapper: anything carrying the ref alongside whatever
// other context the consumer needs.
type OffersFeed struct {
    Ref stream.SourceRef[int]
}

// Both nodes register the wrapper plus the int element type:
//   stream.RemoteOptions(),
//   remote.WithSerializables(new(OffersFeed), new(int)),

// Producer node (sysA)
ref, err := stream.Of(1, 2, 3, 4, 5).SourceRef(ctx, sysA)
if err != nil { return err }

// Hand the ref off through whatever channel makes sense for your app:
// a remote Tell, a cluster pub/sub topic, an actor registry, an HTTP
// response body decoded back into an OffersFeed, etc.
publisher.Tell(ctx, downstreamPID, &OffersFeed{Ref: ref})

// Consumer node (sysB), inside the actor that received OffersFeed:
feed := msg.(*OffersFeed)
col, sink := stream.Collect[int]()
h, err := feed.Ref.Source(sysB).
    Via(stream.Map(func(n int) int { return n * 2 })).
    To(sink).
    Run(ctx, sysB)

The underlying source pipeline is materialised lazily: only when a subscriber connects via ref.Source(...).Run(...). Constructing the ref is cheap. The source pipeline stops when the consumer stops or is aborted, and as soon as the stream ends; a failure of the source reaches the consumer as the stream’s error.

`SinkRef`: publish a consumer

The mirror image: a SinkRef[T] exposes a local Sink[T] so a remote node can pipe data into it.

SinkRef: publish a consumer

// Consumer node (sysA)
col, sink := stream.Collect[int]()
ref, err := sink.SinkRef(ctx, sysA)
if err != nil { return err }

// Producer node (sysB)
h, err := stream.From(stream.Of(10, 20, 30, 40, 50)).
    To(ref.Sink(sysB)).
    Run(ctx, sysB)

Semantics

Single subscription per ref. A second ref.Source(...) / ref.Sink(...) materialisation observes a stream-level error ("already subscribed" while the first stream is active, "already consumed" after it finishes). Refs are one-shot, matching Akka StreamRefs.
Wire-level credit. The consumer’s downstream demand drives streamRequestWire to the producer; the producer ships exactly that many elements before pausing. A bounded pending queue (1024 elements) on the producer side converts a slow-consumer scenario into a visible "backpressure overflow" stream error rather than unbounded buffering on the producer node.
Bounded lifecycle. After the stream completes (success or error), the endpoint actor stays alive for a 30-second grace window so a racing late subscriber gets a clean rejection rather than telling a dead actor and hanging. After the grace window the endpoint reaps itself via system.ScheduleOnce. Refs that are never consumed are reaped at actor-system shutdown.
Death detection. Bridges Watch their remote endpoint as defense in depth: an unexpected endpoint death (panic, system shutdown mid-stream, expired grace on a stale ref) surfaces as a stream error on the bridge’s StreamHandle.Err(), never a hang. Combined with retry-with-jitter resolution that absorbs cluster propagation latency, this keeps cross-node graphs reliable under load.

Reference

Source[T].SourceRef(ctx, sys): Publish a Source[T] as a wire-portable SourceRef[T].
(SourceRef[T]).Source(sys) Source[T]: Adapt a ref back into a Source[T] for use in a local graph.
Sink[T].SinkRef(ctx, sys): Publish a Sink[T] as a wire-portable SinkRef[T].
(SinkRef[T]).Sink(sys) Sink[T]: Adapt a ref back into a Sink[T] for use in a local graph.
RemoteOptions() remote.Option: Registers the control-plane wire protocol with remote.NewConfig.

Backpressure

GoAkt Streams uses credit-based demand propagation. The sink signals how many elements it can accept; each stage propagates that demand upstream. Sources produce only what is requested.

Backpressure

Sink ──[streamRequest n=224]──► Flow ──[streamRequest n=224]──► Source
     ◄─[streamElement × 224]──       ◄─[streamElement × 224]──

Key parameters (in StageConfig):

InitialDemand: First batch of demand sent by a sink to its upstream. Default: 224.
RefillThreshold: Credit level at which a refill is requested. Default: 64.
BufferSize: Mailbox buffer (bounded mailbox capacity). Default: 256.

The sliding-window watermark means:

When credit > RefillThreshold, no refill is sent: the pipeline is healthy.
When credit ≤ RefillThreshold, the stage requests InitialDemand - credit more elements from upstream.
This keeps in-flight data bounded while keeping the pipeline saturated.

Error handling

Each stage can be configured with an independent ErrorStrategy:

FailFast (default): Propagate error downstream, shut down the pipeline.
Resume: Skip the failed element; request a replacement from upstream.
Retry: Re-process the element up to RetryConfig.MaxAttempts times before failing.
Supervise: Delegate to actor supervision (currently equivalent to FailFast).

Error handling

stream.From(stream.FromChannel(ch)).
    Via(stream.TryMap(parse).
        WithErrorStrategy(stream.Retry).
        WithRetryConfig(stream.RetryConfig{MaxAttempts: 3})).
    To(sink).
    Run(ctx, sys)

Overflow strategies

OverflowStrategy has four values:

DropHead: Discard the oldest buffered element.
DropTail (default): Discard the newest incoming element.
BackpressureSource: Hold the producer back until space is available.
FailSource: Terminate the stream with an error.

Only substream buffers act on the strategy today (see SubFlow.WithSubstreamBuffer), and they default to BackpressureSource. Linear stages never overflow: they are bounded by demand, so Buffer and Source.WithOverflowStrategy record the strategy but do not apply it.

Stage fusion

Adjacent stateless stages (Map, Filter) are automatically fused into a single actor, eliminating intermediate mailbox enqueues and reducing allocations.

Stage fusion

// Without fusion (FuseNone): source → map-actor → filter-actor → sink-actor
// With fusion (FuseStateless, default): source → fused-actor → sink-actor

handle, err := graph.
    WithFusion(stream.FuseStateless). // default; explicit here for clarity
    Run(ctx, sys)

FuseStateless (default): Fuses adjacent Map and Filter stages into one actor.
FuseNone: No fusion; each stage is a separate actor (useful for debugging).
FuseAggressive: Currently behaves like FuseStateless.

Fusion is transparent: the external API and semantics are identical. A fused actor always fails fast and calls no tracer, so a stage configured with an ErrorStrategy other than FailFast or with a Tracer is never fused and keeps its own actor. Disable fusion with FuseNone when profiling individual stage throughput.

StreamHandle lifecycle

RunnableGraph.Run returns a StreamHandle for controlling the pipeline at runtime:

StreamHandle lifecycle

handle, err := graph.Run(ctx, sys)
if err != nil {
    return err
}

// Wait for natural completion
<-handle.Done()
if err := handle.Err(); err != nil {
    log.Printf("stream %s failed: %v", handle.ID(), err)
}

// Orderly shutdown: drains in-flight elements before stopping
if err := handle.Stop(ctx); err != nil {
    return err
}

// Immediate abort: discards all buffered elements
handle.Abort()

ID: Unique stream identifier.
Done: Channel closed when the stream terminates (naturally or by error).
Err: Terminal error; valid only after Done is closed.
Stop: Orderly shutdown; drains in-flight elements, then stops all stages.

Stop cancels the source and lets in-flight work drain. A stream whose in-flight work never finishes, such as a FlatMapMerge whose inner sources are endless, does not end on Stop: use Abort.

Abort: Immediate termination; discards buffered elements.
Metrics: Returns a StreamMetrics snapshot.

Observability

Metrics

StreamHandle.Metrics() returns a live StreamMetrics snapshot. The counters are plain struct fields, summed over the source and sink stages:

Metrics

m := handle.Metrics()
log.Printf("in=%d out=%d dropped=%d errors=%d", m.ElementsIn, m.ElementsOut, m.DroppedElements, m.Errors)

Tracer

Attach a Tracer to a Flow for per-element distributed tracing:

Tracer

type myTracer struct{}

func (t *myTracer) OnElement(stageName string, seqNo uint64, latencyNs int64) {
    span.AddEvent(stageName, trace.WithAttributes(
        attribute.Int64("seqNo", int64(seqNo)),
        attribute.Int64("latencyNs", latencyNs),
    ))
}
func (t *myTracer) OnDemand(stageName string, n int64)    {}
func (t *myTracer) OnError(stageName string, err error)   { span.RecordError(err) }
func (t *myTracer) OnComplete(stageName string)            { span.End() }

stream.From(src).
    Via(stream.Map(transform).WithTracer(&myTracer{})).
    To(sink).
    Run(ctx, sys)

The tracer is called by the element-wise flows (Map, TryMap, Filter, FlatMap, Flatten, Buffer, Deduplicate, Scan, WithContext). Source.WithTracer and Sink.WithTracer store the tracer, but source and sink stages do not call it yet.

Configuration reference

StageConfig carries per-stage configuration. Apply it via the builder methods on Flow, Sink, or Source. Builder calls compose: each returns a new stage that keeps the options set before it.

Configuration reference

stream.Map(fn).
    WithErrorStrategy(stream.Retry).
    WithRetryConfig(stream.RetryConfig{MaxAttempts: 3}).
    WithName("parse-stage").
    WithTracer(tracer).
    WithMailbox(actor.NewBoundedMailbox(512))

Flow and Sink offer WithErrorStrategy, WithRetryConfig, WithMailbox, WithName, WithTags and WithTracer; Source offers WithOverflowStrategy and WithTracer. The remaining fields keep their defaults unless an operator sets them (for example Buffer sets InitialDemand, RefillThreshold and BufferSize from its size).

InitialDemand: First demand batch from sink to upstream. Type: int64. Default: 224.
RefillThreshold: Credit level that triggers a refill request. Type: int64. Default: 64.
ErrorStrategy: Element-level failure handling. Type: ErrorStrategy. Default: FailFast.
RetryConfig.MaxAttempts: Retry attempts after the first failed call when ErrorStrategy is Retry. Type: int. Default: 1.
OverflowStrategy: Recorded on the stage; not applied by linear stages (see Overflow strategies). Type: OverflowStrategy. Default: DropTail.
PullTimeout: Timeout for FromActor pull requests. Type: time.Duration. Default: 5s.
BufferSize: Bounded mailbox capacity (BufferSize × 2 slots). Type: int. Default: 256.
Mailbox: Override the default mailbox implementation. Type: actor.Mailbox. Default: nil.
Name: Custom actor name for this stage. Type: string. Default: auto.
Tags: Key/value labels stored on the stage; not yet read by any stage. Type: map[string]string. Default: nil.
Tracer: Optional distributed tracing hook. Type: Tracer. Default: nil.
Fusion: Whether this stage may participate in stage fusion. Type: bool. Default: true.
OnDrop: Called when an element is dropped due to overflow or exhausted retries. Type: func. Default: nil.

Comparison with other approaches

Go channels and goroutines

Go channels are the native concurrency primitive. They are low-level, composable, and require no dependencies: but they demand that the developer hand-craft every concern that GoAkt Streams handles automatically.

Backpressure: Channels + goroutines: Manual (blocking sends or select/default). GoAkt Streams: Automatic credit-based demand propagation.
Error handling: Channels + goroutines: Explicit error channels and coordination logic. GoAkt Streams: ErrorStrategy per stage (FailFast, Resume, Retry).
Fan-out: Channels + goroutines: Manual goroutine fan; complex synchronisation. GoAkt Streams: Broadcast, Balance primitives.
Fan-in: Channels + goroutines: select or sync.WaitGroup coordination. GoAkt Streams: Merge, Combine primitives.
Windowing: Channels + goroutines: Custom timer goroutines and buffers. GoAkt Streams: Batch[T] built-in.
Rate limiting: Channels + goroutines: Custom token bucket / ticker goroutines. GoAkt Streams: Throttle[T] built-in.
Cancellation: Channels + goroutines: context.Context plumbing through every goroutine. GoAkt Streams: handle.Stop / handle.Abort.
Lifecycle: Channels + goroutines: Manual sync.WaitGroup + goroutine tracking. GoAkt Streams: StreamHandle.Done + supervision tree.
Observability: Channels + goroutines: Manual counters. GoAkt Streams: Built-in StreamMetrics and Tracer.

When to prefer channels: simple point-to-point pipelines with one producer and one consumer, or when you need the absolute minimum overhead and are comfortable managing error propagation manually. When to prefer GoAkt Streams: multi-stage transformations, fan-out/fan-in topologies, variable-rate sources, anything requiring windowing, batching, or rate limiting, and whenever you want a managed lifecycle.

RxGo

RxGo is a Go implementation of Reactive Extensions (ReactiveX). It offers a rich operator library and a familiar API for developers coming from RxJS or RxJava.

Backpressure: RxGo: Not supported in v2; the observable push model can overflow without consumer-side control. GoAkt Streams: Pull-only, credit-based; source never outpaces downstream.
Concurrency model: RxGo: Goroutines with optional WithPool options. GoAkt Streams: Actor-per-stage; supervised, lifecycle-managed.
Type safety: RxGo: interface{} items in v2; generic operators not yet complete. GoAkt Streams: Full generics: Flow[In, Out], Source[T], Sink[T].
Actor integration: RxGo: None. GoAkt Streams: First-class: FromActor, ToActor, actor supervision.
Error model: RxGo: Error items in the stream; OnError handler. GoAkt Streams: Typed ErrorStrategy (FailFast, Resume, Retry) per stage.
Lifecycle: RxGo: Dispose pattern. GoAkt Streams: StreamHandle with Stop / Abort / Done.
Operator richness: RxGo: Very rich (debounce, zip, skip, take, window…). GoAkt Streams: Core operators; FlatMap, Batch, Throttle, ParallelMap.
Fan-out: RxGo: Connectable observables. GoAkt Streams: Broadcast and Balance with automatic backpressure.
Supervision: RxGo: None. GoAkt Streams: GoAkt actor supervision tree.
Status: RxGo: Maintenance mode. GoAkt Streams: Actively maintained.

Choose RxGo if you are building standalone Go applications, want the full ReactiveX operator vocabulary, and do not need backpressure or actor integration. Choose GoAkt Streams if you need guaranteed backpressure, are already in a GoAkt system, or require supervised stages with automatic restart.

Benthos / Redpanda Connect

Benthos (now Redpanda Connect) is a production-grade stream-processing engine focused on connecting data systems. It is configuration-driven and ships with hundreds of connectors.

Primary audience: Benthos / Redpanda Connect: Platform / DevOps teams building data pipelines. GoAkt Streams: Go application developers integrating stream logic into services.
Configuration model: Benthos / Redpanda Connect: YAML configuration files. GoAkt Streams: Code-first Go API.
Backpressure: Benthos / Redpanda Connect: Ack-based; built into every connector. GoAkt Streams: Credit-based demand propagation.
Connectors: Benthos / Redpanda Connect: 200+ built-in (Kafka, S3, HTTP, databases…). GoAkt Streams: User-defined via Source, Sink wrappers.
Type safety: Benthos / Redpanda Connect: Untyped message blobs. GoAkt Streams: Generic Source[T] / Flow[In,Out].
Embedding: Benthos / Redpanda Connect: Supported via Go library. GoAkt Streams: Native: it is a Go library.
Processing model: Benthos / Redpanda Connect: Component DAG configured in YAML. GoAkt Streams: Composable Go values assembled in code.
Testing: Benthos / Redpanda Connect: Unit tests via YAML fixtures. GoAkt Streams: Standard Go tests; testkit.
Error handling: Benthos / Redpanda Connect: Retry / DLQ per connector. GoAkt Streams: ErrorStrategy + OnDrop callback per stage.
Actor integration: Benthos / Redpanda Connect: None. GoAkt Streams: First-class GoAkt actor interop.

Choose Benthos if you are building a standalone data pipeline product, need ready-made connectors to external systems, or want configuration-driven deployments without writing Go code. Choose GoAkt Streams if stream processing is one facet of a larger GoAkt application, you want to express pipelines in idiomatic Go code, or you need deep integration with the actor supervision model.

Apache Kafka Streams

Kafka Streams is a JVM client library for building stateful stream processors on top of Apache Kafka.

Language: Kafka Streams: Java / Kotlin (JVM). GoAkt Streams: Go.
Broker dependency: Kafka Streams: Requires Apache Kafka. GoAkt Streams: No broker; in-process.
Durability: Kafka Streams: Persistent topics; replayable. GoAkt Streams: In-memory within the process; ephemeral.
State stores: Kafka Streams: RocksDB-backed state stores; fault-tolerant. GoAkt Streams: Stateful stages via actor state; no persistent store.
Scalability: Kafka Streams: Horizontal via Kafka partitions. GoAkt Streams: Vertical + actor location transparency.
Backpressure: Kafka Streams: Implicit via Kafka consumer offset. GoAkt Streams: Explicit credit-based demand.
Exactly-once: Kafka Streams: Exactly-once semantics with transactions. GoAkt Streams: Exactly-once within a local pipeline.
Windowing: Kafka Streams: Tumbling, hopping, sliding, session windows. GoAkt Streams: Batch[T] for simple windowing.
Table joins: Kafka Streams: KTable / KStream joins with changelog topics. GoAkt Streams: Not built-in; compose via actor state.
Operator richness: Kafka Streams: Filter, map, flatMap, groupBy, aggregate, join. GoAkt Streams: See flow reference table above.

Choose Kafka Streams when you need durable, replayable, horizontally scalable stream processing on the JVM with Kafka as the backbone. Choose GoAkt Streams for in-process stream processing embedded in a Go service, with no external broker dependency and tight integration with GoAkt actor supervision.

Akka Streams

Akka Streams (Scala / Java) is the direct inspiration for GoAkt Streams. Both implement the Reactive Streams specification.

Language: Akka Streams: Scala / Java (JVM). GoAkt Streams: Go.
Backpressure: Akka Streams: Reactive Streams specification. GoAkt Streams: Credit-based pull; equivalent semantics.
Graph DSL: Akka Streams: Rich typed graph builder (GraphDSL). GoAkt Streams: Graph named-node builder.
Stage fusion: Akka Streams: Automatic operator fusion. GoAkt Streams: FuseStateless (default), FuseNone, FuseAggressive.
Materializer: Akka Streams: ActorMaterializer. GoAkt Streams: GoAkt ActorSystem.
Error handling: Akka Streams: SupervisionStrategy per flow. GoAkt Streams: ErrorStrategy per stage.
Fan-out: Akka Streams: Broadcast, Balance, Partition. GoAkt Streams: Broadcast, Balance.
Fan-in: Akka Streams: Merge, MergePreferred, Zip, ZipWith. GoAkt Streams: Merge, Combine.
Substreams: Akka Streams: groupBy, splitWhen, splitAfter, mergeSubstreams, concatSubstreams. GoAkt Streams: GroupBy, SplitWhen, SplitAfter, MergeSubstreams (with per-substream buffer and error strategy).
Cross-node refs: Akka Streams: SourceRef, SinkRef. GoAkt Streams: SourceRef, SinkRef (single-subscription, grace-period reaped, credit-paced).
Actor interop: Akka Streams: Source.actorRef, Sink.actorRef. GoAkt Streams: FromActor, ToActor, ToActorNamed.
Type safety: Akka Streams: Full (Scala generics). GoAkt Streams: Full (Go generics).
Operator richness: Akka Streams: Very rich. GoAkt Streams: Core operators; sufficient for most workloads.
Runtime: Akka Streams: JVM; Akka actor system. GoAkt Streams: Go runtime; GoAkt actor system.
Licence: Akka Streams: BSL 1.1 (Lightbend). GoAkt Streams: MIT.

GoAkt Streams deliberately mirrors Akka Streams’ core concepts: lazy Source, Flow, Sink, demand-driven backpressure, stage fusion: while adapting them to Go idioms and the GoAkt actor hierarchy. The PullRequest / PullResponse actor source protocol is analogous to Akka’s Source.actorRefWithBackpressure.

Observability: Metrics and OpenTelemetry integration
Event Streams: Internal system and cluster event bus
PubSub: Application-level topic-based pub/sub
Actor Scheduling: Timer and cron scheduling used by Tick and Batch
Supervision: Actor failure handling that underpins stream stage recovery

​Why

​Overview

​Core abstractions

​Quick start

​Linear pipeline

​Type-changing flows

​Sources

​Finite sources

​Channel source

​Actor source

​Tick source

​Network source

​Reference

​Flows

​Mapping and filtering

​Streaming flat-map

​Windowing and rate control

​Stateful flows

​Parallel processing

​Reference

​Sinks

​Common sinks

​Actor and channel sinks

​Reference

​Fan-out and fan-in

​Broadcast

​Balance

​Partition

​Unzip

​Merge

​Concat

​Combine (zip, two-input mixed-type)

​Zip / ZipWith (N-input same-type)

​MergePreferred

​MergePrioritized

​MergeLatest

​MergeSequence

​Substreams (SubFlow)

​Splitting on a delimiter

​Cardinality cap

​Per-substream backpressure

​Per-substream error strategy

​Reference

​Pipeline DSL (Graph builder)

​Cross-node stream refs

​Setup

​SourceRef: publish a producer

​SinkRef: publish a consumer

​Semantics

​Reference

​Backpressure

​Error handling

​Overflow strategies

​Stage fusion

​StreamHandle lifecycle

​Observability

​Metrics

​Tracer

​Configuration reference

​Comparison with other approaches

​Go channels and goroutines

​RxGo

​Benthos / Redpanda Connect

​Apache Kafka Streams

​Akka Streams

​Related

Why

Overview

Core abstractions

Quick start

Linear pipeline

Type-changing flows

Sources

Finite sources

Channel source

Actor source

Tick source

Network source

Reference

Flows

Mapping and filtering

Streaming flat-map

Windowing and rate control

Stateful flows

Parallel processing

Reference

Sinks

Common sinks

Actor and channel sinks

Reference

Fan-out and fan-in

Broadcast

Balance

Partition

Unzip

Merge

Concat

Combine (zip, two-input mixed-type)

Zip / ZipWith (N-input same-type)

MergePreferred

MergePrioritized

MergeLatest

MergeSequence

Substreams (SubFlow)

Splitting on a delimiter

Cardinality cap

Per-substream backpressure

Per-substream error strategy

Reference

Pipeline DSL (Graph builder)

Cross-node stream refs

Setup

`SourceRef`: publish a producer

`SinkRef`: publish a consumer

Semantics

Reference

Backpressure

Error handling

Overflow strategies

Stage fusion

StreamHandle lifecycle

Observability

Metrics

Tracer

Configuration reference

Comparison with other approaches

Go channels and goroutines

RxGo

Benthos / Redpanda Connect

Apache Kafka Streams

Akka Streams

Related