Reactive Streams

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]

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

// 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 }),
)

// 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
})

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)

// 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)

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)

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)

// 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]()

// 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)
})

// 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 with overflow strategy
stream.Buffer[int](256, stream.DropTail)

// 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 })

// 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) })

// 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]()

// 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)

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)

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)
}

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)
}

type kv struct { K string; V int }

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

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

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

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

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}
})

// []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),
)

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

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

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)

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, ...
)

// 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)

// 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 })

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

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

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

// 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)

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

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
)

// 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)

// 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)

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

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

// 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)

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()

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

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)

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