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

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

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

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

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

stream.From(stream.Combine(prices, orders, func(p Price, o Order) Quote {
    return Quote{Price: p, Order: o}
})).To(stream.ForEach(func(q Quote) { submit(q) })).
    Run(ctx, sys)

g := stream.NewGraph()

g.Source("in", stream.FromChannel(eventCh))
g.Flow("validate", stream.From("in"), stream.Filter[Event](isValid))
g.Flow("enrich",   stream.From("validate"), stream.Map(enrich))
g.Sink("db",       stream.From("enrich"), stream.ForEach(func(e Event) { db.Insert(e) }))
g.Sink("metrics",  stream.From("enrich"), stream.ForEach(func(e Event) { metrics.Record(e) }))

handle, err := g.Build().Run(ctx, sys)

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