reagin samber/ro reactive streams
Inspect go.mod, imported plugins, source construction, PipeN chains, subscriptions, subjects, sharing, buffers, retries, and shutdown. Verify every operator against the pinned release because the core and plugin surfaces evolve separately.
Define the stream contract
Before changing a pipeline, establish:
- whether each subscription creates a fresh execution or shares one source;
- ordering and callback-concurrency guarantees;
- who owns source goroutines, channels, timers, and external handles;
- completion, terminal error, cancellation, and retry behavior;
- the slow-consumer policy and maximum buffered state;
- what late subscribers receive.
Use a finite loop or task group when it expresses the contract more directly. ro is most useful when values arrive over time and composition, sharing, timing, or terminal-event handling is central.
Choose sources and pipeline APIs
Representative source constructors include Just, FromSlice, FromChannel, Range, Interval, Timer, Defer, Future, and NewObservable. Check callback signatures, context propagation, and whether a source is finite.
Prefer typed Pipe2, Pipe3, and other available PipeN functions when stages change element types. The untyped Pipe surface, where present, trades compile-time checking for dynamic composition. Collect and terminal operators wait for completion, so they are unsuitable for an unbounded source unless the pipeline first terminates it.
Operator families commonly include:
| Need | Representative operators |
|---|---|
| Transform | Map, MapErr, FlatMap, Scan, Reduce |
| Select | Filter, Take*, Skip*, Distinct*, Find |
| Combine | Merge, Concat, Zip*, CombineLatest*, Race |
| Failure | Catch, OnErrorReturn, RetryWithConfig |
| Time and buffering | Timeout, Throttle*, Sample*, Buffer* |
| Teardown visibility | TapOnError, TapOnComplete, TapOnFinalize |
| Context | ContextReset, timeout/deadline helpers, cancellation operators |
Check whether flattening operators preserve order, how many inner streams run concurrently, and when errors cancel siblings.
Cold, shared, and subject sources
A cold observable normally runs per subscription. Share, ShareReplay, and connectable APIs can share a source, with version-specific reset and reference-count behavior.
Subjects are hot bridges from imperative producers. Releases may provide publish, behavior, replay, async, and unicast subjects. Select by late-subscriber and subscriber-count semantics, and bound every replay or unicast buffer. Complete or error a subject through its defined terminal path; terminal state affects later subscribers.
Plugins
Plugins use paths under github.com/samber/ro/plugins/... for concerns such as encoding, HTTP, file events, scheduling, observability, rate limiting, signals, or integrations with other samber modules. Confirm the full import path and module requirement before use; a plugin can introduce its own resource, context, and shutdown contract.
Lifecycle, failure, and pressure
Keep the subscription handle or context cancellation reachable from the component that starts the stream. Source cleanup must run on unsubscribe, context cancellation, error, and normal completion. Close only channels and resources created by that source.
Backpressure is not implied by the reactive abstraction. Choose blocking, bounded buffering, dropping, sampling, batching, or failure from the data contract. Retries need a bound or stopping condition, cancellation, and a source operation safe to repeat.
Verification
Test a cold source with multiple subscriptions and shared/subject behavior with early and late subscribers. Cover values, completion, terminal errors, cancellation, retry exhaustion, buffer limits, slow consumers, ordering, and source cleanup. Use deterministic sources and controllable time instead of narrow sleeps, and run race-enabled tests when callbacks can overlap.