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:

Scenario	Tool	Why
Transform a slice (map, filter, reduce)	INLINECODE2	Finite, synchronous, eager — no stream overhead needed
Simple goroutine fan-out with error handling

errgroup | Standard lib, lightweight, sufficient for bounded concurrency |
| Infinite event stream (WebSocket, tickers, file watcher) | samber/ro | Declarative pipeline with backpressure, retry, timeout, combine |
| Real-time data enrichment from multiple async sources | samber/ro | CombineLatest/Zip compose dependent streams without manual select |
| Pub/sub with multiple consumers sharing one source | samber/ro | Hot observables (Share/Subjects) handle multicast natively |

Key differences: lo vs ro

Aspect	INLINECODE7	INLINECODE8
Data	Finite slices	Infinite streams
Execution

Synchronous, blocking | Asynchronous, non-blocking |
| Evaluation | Eager (allocates intermediate slices) | Lazy (processes items as they arrive) |
| Timing | Immediate | Time-aware (delay, throttle, interval, timeout) |
| Error model | Return (T, error) per call | Error channel propagates through pipeline |
| Use case | Collection transforms | Event-driven, real-time, async pipelines |

Installation

CODEBLOCK0

Core Concepts

Four building blocks:

1. 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), INLINECODE12
3. Operator — a function that transforms an observable into another observable, chained via INLINECODE13
4. Subscription — the connection between observable and observer. Call .Wait() to block or .Unsubscribe() to cancel

CODEBLOCK1

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 with	Behavior
INLINECODE17	Cold → hot with reference counting. Last unsubscribe tears down
INLINECODE18

Same as Share + buffers last N values for late subscribers |
| Connectable() | Cold → hot, but waits for explicit .Connect() call |
| Subjects | Natively hot — call .Send(), .Error(), .Complete() directly |

Subject	Constructor	Replay behavior
INLINECODE24	INLINECODE25	None — late subscribers miss past events
INLINECODE26

NewBehaviorSubject[T](initial) | Replays last value to new subscribers |
| ReplaySubject | NewReplaySubject[T](bufferSize) | Replays last N values |
| AsyncSubject | NewAsyncSubject[T]() | Emits only last value, only on complete |
| UnicastSubject | NewUnicastSubject[T](bufferSize) | Single subscriber only |

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

Operator Quick Reference

Category	Key operators	Purpose
Creation	INLINECODE34, FromSlice, FromChannel, Range, Interval, Defer, INLINECODE40	Create observables from various sources
Transform

Map, MapErr, FlatMap, Scan, Reduce, GroupBy | Transform or accumulate stream values | | Filter | Filter, Take, TakeLast, Skip, Distinct, Find, First, Last | Selectively emit values | | Combine | Merge, Concat, Zip2–Zip6, CombineLatest2–CombineLatest5, Race | Merge multiple observables | | Error | Catch, OnErrorReturn, OnErrorResumeNextWith, Retry, RetryWithConfig | Recover from errors | | Timing | Delay, DelayEach, Timeout, ThrottleTime, SampleTime, BufferWithTime | Control emission timing | | Side effect | Tap/Do, TapOnNext, TapOnError, TapOnComplete | Observe without altering stream | | Terminal | Collect, ToSlice, ToChannel, ToMap | Consume 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

Mistake	Why it fails	Fix
Using ro.OnNext() without error handler	Errors are silently dropped — bugs hide in production	Use ro.NewObserver(onNext, onError, onComplete) with all 3 callbacks
Using untyped Pipe() instead of Pipe2/INLINECODE91

Loses compile-time type safety, errors surface at runtime | Use Pipe2, Pipe3...Pipe25 for typed operator chains | | Forgetting .Unsubscribe() on infinite streams | Goroutine leak — the observable runs forever | Use TakeUntil(signal), context cancellation, or explicit Unsubscribe() | | Using Share() when cold is sufficient | Unnecessary complexity, harder to reason about lifecycle | Use hot observables only when multiple consumers need the same stream | | Using samber/ro for finite slice transforms | Stream overhead (goroutines, subscriptions) for a synchronous operation | Use samber/lo — it's simpler, faster, and purpose-built for slices | | Not propagating context for cancellation | Streams ignore shutdown signals, causing resource leaks on termination | Chain ContextWithTimeout or ThrowOnContextCancel in the pipeline |

Best Practices

1. 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:

Category	Plugins	Import path prefix
Encoding	JSON, CSV, Base64, Gob	INLINECODE121
Network

HTTP, I/O, FSNotify | plugins/http, plugins/io, plugins/fsnotify |
| Scheduling | Cron, ICS | plugins/cron, plugins/ics |
| Observability | Zap, Slog, Zerolog, Logrus, Sentry, Oops | plugins/observability/..., plugins/samber/oops |
| Rate limiting | Native, Ulule | plugins/ratelimit/... |
| Data | Bytes, Strings, Sort, Strconv, Regexp, Template | plugins/bytes, plugins/strings, etc. |
| System | Process, Signal | plugins/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