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: Reason as thoroughly as possible when designing advanced reactive pipelines or choosing between cold/hot observables, subjects, and combining operators — wrong architecture leads to resource leaks or missed events. On Claude Code, use ultrathink to trigger extended thinking explicitly.

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. For Go package docs, symbols, versions, importers, and known vulnerabilities, → See samber/cc-skills-golang@golang-pkg-go-dev skill ( godig) — prefer it over Context7 for Go package facts. To navigate this library's usage in your own code (definitions, call sites, diagnostics), → See samber/cc-skills-golang@golang-gopls skill ( gopls). Context7 remains a fallback for docs not indexed on pkg.go.dev.

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:

ScenarioToolWhyTransform a slice (map, filter, reduce)samber/loFinite, synchronous, eager — no stream overhead neededSimple goroutine fan-out with error handlingerrgroupStandard lib, lightweight, sufficient for bounded concurrencyInfinite event stream (WebSocket, tickers, file watcher)samber/roDeclarative pipeline with backpressure, retry, timeout, combineReal-time data enrichment from multiple async sourcessamber/roCombineLatest/Zip compose dependent streams without manual selectPub/sub with multiple consumers sharing one sourcesamber/roHot observables (Share/Subjects) handle multicast natively

Key differences: lo vs ro

Aspectsamber/lo``samber/roDataFinite slicesInfinite streamsExecutionSynchronous, blockingAsynchronous, non-blockingEvaluationEager (allocates intermediate slices)Lazy (processes items as they arrive)TimingImmediateTime-aware (delay, throttle, interval, timeout)Error modelReturn (T, error) per callError channel propagates through pipelineUse caseCollection transformsEvent-driven, real-time, async pipelines

Installation#

bash

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

go

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.

Convert withBehaviorShare()Cold → hot with reference counting. Last unsubscribe tears downShareReplay(n)Same as Share + buffers last N values for late subscribersConnectable()Cold → hot, but waits for explicit .Connect() callSubjectsNatively hot — call .Send(), .Error(), .Complete() directlySubjectConstructorReplay behaviorPublishSubject``NewPublishSubject[T]()None — late subscribers miss past eventsBehaviorSubject``NewBehaviorSubject[T](initial)Replays last value to new subscribersReplaySubject``NewReplaySubject[T](bufferSize)Replays last N valuesAsyncSubject``NewAsyncSubject[T]()Emits only last value, only on completeUnicastSubject``NewUnicastSubject[T](bufferSize)Single subscriber only

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

Operator Quick Reference#

CategoryKey operatorsPurposeCreationJust, FromSlice, FromChannel, Range, Interval, Defer, FutureCreate observables from various sourcesTransformMap, MapErr, FlatMap, Scan, Reduce, GroupByTransform or accumulate stream valuesFilterFilter, Take, TakeLast, Skip, Distinct, Find, First, LastSelectively emit valuesCombineMerge, Concat, Zip2– Zip6, CombineLatest2– CombineLatest5, RaceMerge multiple observablesErrorCatch, OnErrorReturn, OnErrorResumeNextWith, Retry, RetryWithConfigRecover from errorsTimingDelay, DelayEach, Timeout, ThrottleTime, SampleTime, BufferWithTimeControl emission timingSide effectTap/ Do, TapOnNext, TapOnError, TapOnCompleteObserve without altering streamTerminalCollect, ToSlice, ToChannel, ToMapConsume stream into Go types

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#

MistakeWhy it failsFixUsing ro.OnNext() without error handlerErrors are silently dropped — bugs hide in productionUse ro.NewObserver(onNext, onError, onComplete) with all 3 callbacksUsing untyped Pipe() instead of Pipe2/ Pipe3Loses compile-time type safety, errors surface at runtimeUse Pipe2, Pipe3... Pipe25 for typed operator chainsForgetting .Unsubscribe() on infinite streamsGoroutine leak — the observable runs foreverUse TakeUntil(signal), context cancellation, or explicit Unsubscribe()Using Share() when cold is sufficientUnnecessary complexity, harder to reason about lifecycleUse hot observables only when multiple consumers need the same streamUsing samber/ro for finite slice transformsStream overhead (goroutines, subscriptions) for a synchronous operationUse samber/lo — it's simpler, faster, and purpose-built for slicesNot propagating context for cancellationStreams ignore shutdown signals, causing resource leaks on terminationChain ContextWithTimeout or ThrowOnContextCancel in the pipeline

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:

CategoryPluginsImport path prefixEncodingJSON, CSV, Base64, Gobplugins/encoding/...NetworkHTTP, I/O, FSNotifyplugins/http, plugins/io, plugins/fsnotifySchedulingCron, ICSplugins/cron, plugins/icsObservabilityZap, Slog, Zerolog, Logrus, Sentry, Oopsplugins/observability/..., plugins/samber/oopsRate limitingNative, Ululeplugins/ratelimit/...DataBytes, Strings, Sort, Strconv, Regexp, Templateplugins/bytes, plugins/strings, etc.SystemProcess, Signalplugins/proc, plugins/signal

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