Golang Samber Ro

Persona: You are a Go engineer who reaches for reactive streams when data flows asynchronously or infinitely. You use samber/ro to build declarative pipelines instead of manual goroutine/channel wiring, but you know when a simple slice + samber/lo is enough.

Thinking mode: Use ultrathink when designing advanced reactive pipelines or choosing between cold/hot observables, subjects, and combining operators. Wrong architecture leads to resource leaks or missed events.

samber/ro — Reactive Streams for Go

Go implementation of ReactiveX. Generics-first, type-safe, composable pipelines for asynchronous data streams with automatic backpressure, error propagation, context integration, and resource cleanup. 150+ operators, 5 subject types, 40+ plugins.

Official Resources:

• github.com/samber/ro
• ro.samber.dev
• pkg.go.dev/github.com/samber/ro

This skill is not exhaustive. Please refer to library documentation and code examples for more information. Context7 can help as a discoverability platform.

Why samber/ro (Streams vs Slices)

Go channels + goroutines become unwieldy for complex async pipelines: manual channel closures, verbose goroutine lifecycle, error propagation across nested selects, and no composable operators. samber/ro solves this with declarative, chainable stream operators.

When to use which tool:

Key differences: lo vs ro

Installation

go get github.com/samber/ro

Core Concepts

Four building blocks:

1. Observable — a data source that emits values over time. Cold by default: each subscriber triggers independent execution from scratch 2. Observer — a consumer with three callbacks: onNext(T), onError(error), onComplete() 3. Operator — a function that transforms an observable into another observable, chained via Pipe 4. Subscription — the connection between observable and observer. Call .Wait() to block or .Unsubscribe() to cancel

observable := ro.Pipe2(
    ro.RangeWithInterval(0, 5, 1*time.Second),
    ro.Filter(func(x int) bool { return x%2 == 0 }),
    ro.Map(func(x int) string { return fmt.Sprintf("even-%d", x) }),
)

observable.Subscribe(ro.NewObserver(
    func(s string) { fmt.Println(s) },      // onNext
    func(err error) { log.Println(err) },    // onError
    func() { fmt.Println("Done!") },         // onComplete
))
// Output: "even-0", "even-2", "even-4", "Done!"

// Or collect synchronously:
values, err := ro.Collect(observable)

Cold vs Hot Observables

Cold (default): each .Subscribe() starts a new independent execution. Safe and predictable — use by default.

Hot: multiple subscribers share a single execution. Use when the source is expensive (WebSocket, DB poll) or subscribers must see the same events.

For subject details and hot observable patterns, see Subjects Guide.

Operator Quick Reference

Use typed Pipe2, Pipe3 ... Pipe25 for compile-time type safety across operator chains. The untyped Pipe uses any and loses type checking.

For the complete operator catalog (150+ operators with signatures), see Operators Guide.

Common Mistakes

Best Practices

1. Always handle all three events — use NewObserver(onNext, onError, onComplete), not just OnNext. Unhandled errors cause silent data loss 2. Use `Collect()` for synchronous consumption — when the stream is finite and you need []T, Collect blocks until complete and returns the slice + error 3. Prefer typed Pipe functions — Pipe2, Pipe3...Pipe25 catch type mismatches at compile time. Reserve untyped Pipe for dynamic operator chains 4. Bound infinite streams — use Take(n), TakeUntil(signal), Timeout(d), or context cancellation. Unbounded streams leak goroutines 5. Use `Tap`/`Do` for observability — log, trace, or meter emissions without altering the stream. Chain TapOnError for error monitoring 6. Prefer `samber/lo` for simple transforms — if the data is a finite slice and you need Map/Filter/Reduce, use lo. Reach for ro when data arrives over time, from multiple sources, or needs retry/timeout/backpressure

Plugin Ecosystem

40+ plugins extend ro with domain-specific operators:

For the full plugin catalog with import paths and usage examples, see Plugin Ecosystem.

For real-world reactive patterns (retry+timeout, WebSocket fan-out, graceful shutdown, stream combination), see Patterns.

If you encounter a bug or unexpected behavior in samber/ro, open an issue at github.com/samber/ro/issues.

Cross-References

• → See samber/cc-skills-golang@golang-samber-lo skill for finite slice transforms (Map, Filter, Reduce, GroupBy) — use lo when data is already in a slice
• → See samber/cc-skills-golang@golang-samber-mo skill for monadic types (Option, Result, Either) that compose with ro pipelines
• → See samber/cc-skills-golang@golang-samber-hot skill for in-memory caching (also available as an ro plugin)
• → See samber/cc-skills-golang@golang-concurrency skill for goroutine/channel patterns when reactive streams are overkill
• → See samber/cc-skills-golang@golang-observability skill for monitoring reactive pipelines in production

Operators Guide

samber/ro provides 150+ operators organized by category. All operators are generic functions that take and return Observable[T], designed for chaining via Pipe.

Pipeline Construction

// Typed pipes (Pipe1 through Pipe25) — compile-time type safety
result := ro.Pipe2(source, op1, op2)
result := ro.Pipe3(source, op1, op2, op3)

// Untyped pipe — uses any, loses type checking
result := ro.Pipe(source, op1, op2, op3)

// Reusable operator composition (curried)
transform := ro.PipeOp2(op1, op2) // returns func(Observable[A]) Observable[C]
result := transform(source)

// Synchronous collection
values, err := ro.Collect(observable)

Creation Operators

Create observables from various sources. Typically the first argument to Pipe.

// Custom observable with direct control
obs := ro.NewObservable[int](func(ctx context.Context, observer ro.Observer[int]) error {
    observer.Next(1)
    observer.Next(2)
    observer.Complete()
    return nil
})

Transformation Operators

Transform each value in the stream.

// FlatMap: for each user ID, fetch their orders (returns observable)
orders := ro.Pipe2(
    userIDs,
    ro.FlatMap(func(id int) ro.Observable[Order] {
        return fetchOrders(id)
    }),
    ro.Filter(func(o Order) bool { return o.Status == "paid" }),
)

// Scan: running sum
ro.Pipe1(
    ro.Just(1, 2, 3, 4, 5),
    ro.Scan(func(acc, x int) int { return acc + x }, 0),
)
// Emits: 1, 3, 6, 10, 15

Filtering Operators

Selectively emit values from the stream.

All filtering operators have I (indexed), WithContext, and IWithContext variants.

Combining Operators

Merge multiple observables into one.

// Zip: pair user with their settings
ro.Zip2(userStream, settingsStream)
// Emits: lo.Tuple2[User, Settings]

// CombineLatest: re-emit whenever either changes
ro.CombineLatest2(priceStream, quantityStream)
// Emits latest (price, quantity) pair each time either updates

Math and Aggregation

Error Handling

// RetryConfig for exponential backoff
ro.RetryWithConfig[Response](ro.RetryConfig{
    Max:               3,
    Delay:             time.Second,
    BackoffMultiplier: 2.0,
    MaxDelay:          10 * time.Second,
})

Timing and Buffering

Side Effects (Tap / Do)

Execute code without changing the stream. Do is an alias for Tap.

All have WithContext variants. Do, DoOnNext, DoOnError, DoOnComplete, DoOnSubscribe, DoOnFinalize are aliases.

Connectable and Sharing

// ShareConfig controls lifecycle reset
ro.ShareConfig[T]{
    ResetOnComplete:       true,  // re-subscribe on complete
    ResetOnError:          true,  // re-subscribe on error
    ResetOnReferenceCount: true,  // re-subscribe when count drops to 0
}

Context Operators

Conditional Operators

Terminal Operators

Observer and Subscription

// Full observer (recommended)
observer := ro.NewObserver[T](onNext, onError, onComplete)

// Context-aware observer
observer := ro.NewObserverWithContext[T](onNext, onError, onComplete)

// Convenience shortcuts
observer := ro.OnNext[T](func(v T) { ... })
observer := ro.PrintObserver[T]()  // debug: prints all events
observer := ro.NoopObserver[T]()   // discard all events

// Subscription lifecycle
sub := observable.Subscribe(observer)
sub.Wait()          // block until complete or error
sub.Unsubscribe()   // cancel and cleanup
sub.IsActive()      // check if still running
sub.GetError()      // get terminal error

Scheduling

Concurrency Modes

// Safe (default): synchronized emissions
ro.NewObservable[T](fn)
ro.NewSafeObservable[T](fn)

// Unsafe: no synchronization, caller must guarantee single-goroutine access
ro.NewUnsafeObservable[T](fn)

// Eventually safe: allows brief unsynchronized period, then synchronizes
ro.NewEventuallySafeObservable[T](fn)

Reactive Patterns

Real-world patterns for building production reactive pipelines with samber/ro.

Pattern 1: Remote Call with Retry and Timeout

Wrap a remote call (HTTP, gRPC, database) with automatic retry, exponential backoff, timeout, and fallback.

result := ro.Pipe3(
    fetchUser(userID),  // ro.Observable[User] — wraps your remote call
    ro.Timeout[User](5*time.Second),
    ro.RetryWithConfig[User](ro.RetryConfig{
        Max:               3,
        Delay:             500 * time.Millisecond,
        BackoffMultiplier: 2.0,
        MaxDelay:          5 * time.Second,
    }),
    ro.Catch[User](func(err error) ro.Observable[User] {
        log.Printf("remote call failed after retries: %v, using cache", err)
        return getCachedUser(userID)
    }),
)

user, err := ro.Collect(result)

Why ro over plain calls: declarative retry + timeout + fallback in 10 lines vs manual for-loops with sleep, context, and error tracking.

Pattern 2: Continuous Event Stream (Hot Observable)

Share a single long-lived connection (WebSocket, SSE, message queue) across multiple consumers.

// Cold observable wrapping any event stream source
eventStream := ro.NewObservable[TickerEvent](func(ctx context.Context, obs ro.Observer[TickerEvent]) error {
    // connect to your stream source (WebSocket, NATS, Kafka, etc.)
    for {
        event, err := streamSource.Read(ctx)
        if err != nil {
            return err
        }
        obs.Next(event)
    }
})

// Share: one connection, multiple consumers
shared := ro.Pipe1(eventStream, ro.Share[TickerEvent]())

// Consumer 1: update UI
shared.Subscribe(ro.OnNext(func(e TickerEvent) {
    updateDashboard(e)
}))

// Consumer 2: record metrics
shared.Subscribe(ro.OnNext(func(e TickerEvent) {
    metrics.RecordTick(e.Symbol, e.Price)
}))

// Consumer 3: alert on threshold
ro.Pipe1(shared, ro.Filter(func(e TickerEvent) bool {
    return e.Price > alertThreshold
})).Subscribe(ro.OnNext(sendAlert))

The rohttp plugin provides WebSocket and HTTP streaming observables (see Plugin Ecosystem).

Pattern 3: Fan-In from Multiple Sources

Merge events from multiple independent sources, batch, and process.

combined := ro.Pipe2(
    ro.Merge(
        apiStream,
        pushStream,
        cronScheduleStream,
    ),
    ro.Distinct[Event](),
    ro.BufferWithTimeOrCount[Event](100, 5*time.Second),
    ro.Map(func(batch []Event) ProcessResult {
        return processBatch(batch)
    }),
)

When to use Merge vs Concat vs Zip:

Pattern 4: Dependent Data Combination

Combine data from multiple async sources that depend on each other.

// Fetch user and their orders in parallel, combine
profile := ro.Pipe1(
    ro.CombineLatest2(
        fetchUser(userID),
        fetchOrders(userID),
    ),
    ro.Map(func(pair lo.Tuple2[User, []Order]) UserProfile {
        return UserProfile{
            User:   pair.A,
            Orders: pair.B,
        }
    }),
)

For independent data where you need exactly one value from each:

// Zip: waits for one value from each, pairs them
configAndData := ro.Zip2(loadConfig(), loadData())

Pattern 5: Running Aggregation with Scan

Maintain running state across stream values — useful for dashboards, analytics, monitoring.

type Stats struct {
    Count int
    Sum   float64
    Avg   float64
    Max   float64
}

statsStream := ro.Pipe2(
    metricsStream,
    ro.Scan(func(acc Stats, v float64) Stats {
        acc.Count++
        acc.Sum += v
        acc.Avg = acc.Sum / float64(acc.Count)
        if v > acc.Max {
            acc.Max = v
        }
        return acc
    }, Stats{}),
    ro.SampleTime[Stats](5*time.Second), // emit stats every 5s
)

Scan vs Reduce: Scan emits every intermediate state (good for live dashboards). Reduce emits only the final accumulated value (good for batch summaries).

Pattern 6: Error Recovery Cascade

Layer multiple error recovery strategies.

resilient := ro.Pipe3(
    primaryDataSource,
    // Strategy 1: retry transient failures
    ro.RetryWithConfig[Data](ro.RetryConfig{
        Max:   2,
        Delay: time.Second,
    }),
    // Strategy 2: fall back to secondary source
    ro.Catch[Data](func(err error) ro.Observable[Data] {
        log.Warn("primary failed, trying secondary", "err", err)
        return secondaryDataSource
    }),
    // Strategy 3: return cached/default value
    ro.OnErrorReturn[Data](cachedDefault),
)

Order matters: retry first (transient errors), then fallback source (persistent errors), then default value (total failure).

Pattern 7: File System Watcher

React to file changes with debouncing.

import rofsnotify "github.com/samber/ro/plugins/fsnotify"

watcher := ro.Pipe3(
    rofsnotify.Watch("/etc/app/config/"),
    ro.Filter(func(e fsnotify.Event) bool {
        return e.Op&fsnotify.Write != 0
    }),
    ro.ThrottleTime[fsnotify.Event](2*time.Second), // debounce rapid saves
    ro.Map(func(e fsnotify.Event) Config {
        return reloadConfig(e.Name)
    }),
)

watcher.Subscribe(ro.NewObserver(
    func(cfg Config) { applyConfig(cfg) },
    func(err error) { log.Error("config watch failed", "err", err) },
    func() { log.Info("config watcher stopped") },
))

Pattern 8: Graceful Shutdown

Use context or signal observable to cleanly terminate infinite streams.

import rosignal "github.com/samber/ro/plugins/signal"

// Method 1: OS signal
shutdown := rosignal.Notify(syscall.SIGTERM, syscall.SIGINT)

sub := ro.Pipe1(
    workStream,
    ro.TakeUntil[Work, os.Signal](shutdown),
).Subscribe(ro.NewObserver(
    processWork,
    handleError,
    func() { log.Info("gracefully stopped") },
))

sub.Wait() // blocks until SIGTERM/SIGINT

// Method 2: Context cancellation
ctx, cancel := context.WithCancel(context.Background())

sub := ro.Pipe2(
    workStream,
    ro.ContextReset[Work](ctx),
    ro.ThrowOnContextCancel[Work](),
).Subscribe(worker)

// Later: cancel() triggers clean shutdown

Pattern 9: Event-Driven Pipeline with Logging

Full production pipeline with observability at each stage.

import roslog "github.com/samber/ro/plugins/observability/slog"

pipeline := ro.Pipe5(
    eventSource,
    ro.TapOnSubscribe[Event](func() {
        slog.Info("pipeline started")
    }),
    ro.Filter(func(e Event) bool { return e.Valid() }),
    roslog.TapOnNext[Event](logger, slog.LevelDebug), // log each event
    ro.Map(enrichEvent),
    ro.BufferWithTimeOrCount[EnrichedEvent](50, 10*time.Second),
    ro.MapErr(func(batch []EnrichedEvent) (Result, error) {
        return persistBatch(batch)
    }),
    ro.TapOnError[Result](func(err error) {
        slog.Error("pipeline error", "err", err)
        metrics.IncrCounter("pipeline.errors", 1)
    }),
    ro.RetryWithConfig[Result](ro.RetryConfig{Max: 3, Delay: time.Second}),
)

Plugin Ecosystem

samber/ro ships 40+ plugins that extend the core library with domain-specific operators. Plugins are separate Go modules — install only what you need.

go get github.com/samber/ro/plugins/<category>/<name>

Data Manipulation

Encoding and Serialization

import rojson "github.com/samber/ro/plugins/encoding/json"

// Parse JSON stream: []byte -> MyStruct
parsed := ro.Pipe1(rawBytes, rojson.Unmarshal[MyStruct]())

Scheduling

import rocron "github.com/samber/ro/plugins/cron"

// Emit every day at midnight
daily := rocron.Schedule("0 0 * * *")

Network and I/O

import rofsnotify "github.com/samber/ro/plugins/fsnotify"

// Watch directory for changes
events := rofsnotify.Watch("/var/log/app/")
ro.Pipe1(events, ro.Filter(func(e fsnotify.Event) bool {
    return e.Op == fsnotify.Write
})).Subscribe(ro.OnNext(func(e fsnotify.Event) {
    log.Println("Modified:", e.Name)
}))

Observability and Logging

import roslog "github.com/samber/ro/plugins/observability/slog"

// Log all stream events via slog
ro.Pipe1(
    dataStream,
    roslog.Tap[Data](logger, slog.LevelInfo),
)

Rate Limiting

Text Processing

System Integration

import rosignal "github.com/samber/ro/plugins/signal"

// Observable that emits on SIGTERM/SIGINT
shutdown := rosignal.Notify(syscall.SIGTERM, syscall.SIGINT)

// Use as TakeUntil signal for graceful shutdown
ro.Pipe1(workStream, ro.TakeUntil[Work, os.Signal](shutdown))

Validation

Testing

Utilities

Plugin Design Convention

All plugins follow the same pattern:

1. Import the plugin package 2. Use plugin-provided operators in Pipe chains 3. Plugin operators return func(Observable[T]) Observable[R] — standard operator signature 4. No global state — each operator instance is independent

Plugins are documented individually in their package directories. API details for each plugin are available at pkg.go.dev/github.com/samber/ro/plugins/....

Subjects Guide

Subjects are both Observable and Observer — they can receive values (via Send, Error, Complete) and be subscribed to. Subjects are natively hot: subscribers share a single execution, and late subscribers only see future emissions (unless replay is configured).

When to Use Subjects vs Cold Observables

Subject Types

PublishSubject

Standard multicast. Subscribers only see values emitted after they subscribe.

subject := ro.NewPublishSubject[string]()

// Subscriber 1
subject.Subscribe(ro.OnNext(func(s string) {
    fmt.Println("sub1:", s)
}))

subject.Send("hello")  // sub1 sees this

// Subscriber 2 (late)
subject.Subscribe(ro.OnNext(func(s string) {
    fmt.Println("sub2:", s)
}))

subject.Send("world")  // both see this
subject.Complete()

Use when: broadcasting events where late subscribers don't need history — UI events, log streams, notifications.

BehaviorSubject

Replays the last emitted value (or the initial value) to every new subscriber immediately on subscription.

subject := ro.NewBehaviorSubject[int](0) // initial value = 0

// Subscriber 1 immediately receives 0
subject.Subscribe(ro.OnNext(func(v int) {
    fmt.Println("sub1:", v) // 0, then 42
}))

subject.Send(42)

// Subscriber 2 immediately receives 42 (latest value)
subject.Subscribe(ro.OnNext(func(v int) {
    fmt.Println("sub2:", v) // 42
}))

Use when: subscribers need the current state — config values, connection status, latest price.

ReplaySubject

Buffers the last N values and replays them to every new subscriber.

subject := ro.NewReplaySubject[string](3) // buffer size = 3

subject.Send("a")
subject.Send("b")
subject.Send("c")
subject.Send("d") // "a" evicted from buffer

// Late subscriber receives "b", "c", "d" (last 3)
subject.Subscribe(ro.OnNext(func(s string) {
    fmt.Println(s)
}))

Use when: late subscribers need recent history — chat messages, recent logs, last N stock ticks.

AsyncSubject

Emits only the last value and only when the subject completes. If the subject errors, no value is emitted.

subject := ro.NewAsyncSubject[int]()

subject.Subscribe(ro.NewObserver(
    func(v int) { fmt.Println(v) },    // receives 3 only
    func(err error) { },
    func() { fmt.Println("done") },
))

subject.Send(1)
subject.Send(2)
subject.Send(3)
subject.Complete() // triggers emission of 3, then "done"

Use when: only the final result matters — computation result, last response in a batch.

UnicastSubject

Allows exactly one subscriber. Buffers values internally until that subscriber connects.

subject := ro.NewUnicastSubject[int](100) // buffer size

subject.Send(1) // buffered
subject.Send(2) // buffered

// Single subscriber receives buffered + future values
subject.Subscribe(ro.OnNext(func(v int) {
    fmt.Println(v) // 1, 2, then future values
}))
// Second subscribe would panic or error

Use when: single consumer with buffering — job queues, request pipelines where exactly one handler processes events.

Cold to Hot Conversion

When you have a cold observable (e.g. an HTTP request) but need multiple subscribers to share it:

Share

// Each subscriber to `cold` would trigger a separate HTTP request
cold := httpPlugin.Get[Data](url)

// Share: single execution, multiple subscribers
hot := ro.Pipe1(cold, ro.Share[Data]())

hot.Subscribe(uiObserver)       // shares one HTTP call
hot.Subscribe(metricsObserver)  // same data, no extra request

Share uses reference counting: the source subscribes when the first subscriber arrives and unsubscribes when the last one leaves.

ShareReplay

// Late subscribers get the last N values + future values
hot := ro.Pipe1(cold, ro.ShareReplay[Data](1))

Connectable Observable

For precise control over when the shared subscription starts:

connectable := ro.Connectable[Data](cold)

// Set up subscribers first
connectable.Subscribe(observer1)
connectable.Subscribe(observer2)

// Start the shared execution explicitly
sub, err := connectable.Connect(ctx)

Subject Decision Table

Common Subject Mistakes