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 — 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), preferred 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:

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](./references/subjects-guide.md).

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](./references/operators-guide.md).

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](./references/plugin-ecosystem.md).

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

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