원본 저장소의 제목, 예시, 코드, 표, 링크, 이미지를 유지해 표시합니다.
Persona: You are a Go engineer who reaches for reactive streams when data arrives asynchronously or without end. You build pipelines with operators instead of raw goroutine/channel plumbing, and you know when a slice plus lo is the simpler answer.
Modes:
- Build — expressing an event pipeline as an observable graph.
- Review — auditing pipelines for unhandled errors, unbounded streams, and hot/cold confusion.
- Debug — tracing a missed event or a leaked subscription.
When to use: any task centered on samber/ro. For finite collection transforms use samber/lo; for bounded goroutine fan-out the stdlib errgroup may be all you need. Load golang-concurrency and golang-observability alongside when lifecycle and monitoring matter.
Streams vs slices
| Need | Tool |
|---|---|
| Transform a slice once | samber/lo — eager, synchronous |
| Bounded fan-out with error handling | errgroup |
| Infinite event streams (websocket, ticks, fsnotify) | samber/ro |
| Combine several async sources with timing | samber/ro (combine/zip operators) |
| One source, many consumers | samber/ro hot observables / subjects |
The four building blocks
- Observable — emits values over time; cold by default (each subscriber triggers its own execution).
- Observer — consumes the stream through
onNext,onError,onComplete. - Operator — transforms an observable into another, composed by
Pipe. - Subscription — the wiring between them;
Waitblocks,Unsubscribecancels.
odds := ro.Pipe2(
ro.FromChannel(rawCh),
ro.Filter(func(n int) bool { return n%2 != 0 }),
ro.Map(func(n int) string { return fmt.Sprintf("odd-%d", n) }),
)
odds.Subscribe(ro.NewObserver(
func(s string) { out <- s },
func(err error) { log.Printf("stream: %v", err) },
func() { close(out) },
))Typed vs untyped pipelines
Use the typed Pipe2 … Pipe25 family for compile-time chain checking. The bare Pipe takes any and drops type safety — reach for it only when composing operators dynamically.
Cold and hot sources
Cold (the default) re-runs per subscriber: safe, deterministic, but wasteful when the source is expensive. Hot sources share one execution among all subscribers — the natural shape for websockets, DB polls, and any single event source with many consumers.
| Conversion | Behavior |
|---|---|
Share() | cold → hot, reference-counted teardown |
ShareReplay(n) | hot + replays last n values to late joiners |
Connectable() | hot, but waits for an explicit Connect() |
| Subjects | natively hot; you push with Send/Error/Complete |
| Subject | Replay to late subscribers |
|---|---|
PublishSubject | none |
BehaviorSubject | last value |
ReplaySubject | last N values |
AsyncSubject | last value, and only after completion |
UnicastSubject | only the single subscriber |
Subject details and hot-source patterns are in subjects guide.
Operators at a glance
| Family | Key operators |
|---|---|
| Creation | Just, FromSlice, FromChannel, Range, Interval, Defer, Future |
| Transform | Map, MapErr, FlatMap, Scan, Reduce, GroupBy |
| Filter | Filter, Take, Skip, Distinct, First, Last, Find |
| Combine | Merge, Concat, Zip2…Zip6, CombineLatest2…5, Race |
| Error | Catch, OnErrorReturn, Retry, RetryWithConfig |
| Timing | Delay, Timeout, ThrottleTime, SampleTime, BufferWithTime |
| Side effect | Tap, TapOnNext, TapOnError, TapOnComplete |
| Terminal | Collect, ToSlice, ToChannel, ToMap |
The full catalog lives in operators guide.
Common mistakes
| Mistake | Why it fails | Fix |
|---|---|---|
| Subscribing without an error callback | errors vanish silently | NewObserver(onNext, onError, onComplete) |
Bare Pipe | type mismatches surface at runtime | typed Pipe2…Pipe25 |
| No unsubscribe on infinite streams | goroutine leak for life | TakeUntil, context cancellation, explicit Unsubscribe |
Share() when cold suffices | lifecycle complexity for no consumers | hot only when many need the same source |
| Streams for slice work | goroutine + subscription overhead for a sync op | samber/lo |
| Not wiring context | streams ignore shutdown signals | ContextWithTimeout / ThrowOnContextCancel |
Plugins
30–40+ plugins fold domain operators into pipelines — encoding (JSON, CSV), network (plugins/http, plugins/fsnotify), scheduling (plugins/cron), observability (slog, zap, zerolog), rate limiting, and string/data helpers. Browse plugin ecosystem and real-world recipes in patterns.
Best practices
- Always pass all three callbacks; silent stream errors are production time-bombs.
- Favor typed
Pipefamilies; reservePipefor dynamic chains. - Bound every infinite stream —
Take(n),TakeUntil(signal),Timeout(d), or context. - Use
Collectfor finite streams you want as[]T. lofor data you already hold;rofor data that arrives over time.- Report upstream bugs at samber/ro issues.
Cross-references
golang-samber-lo— eager transforms of finite slices.golang-samber-mo— monadic values that thread through pipeline operators.golang-samber-hot— in-memory caching, also shipped as anroplugin.golang-concurrency— goroutine/channel patterns when streams are overkill.golang-observability— tracing and metrics for reactive pipelines.golang-pkg-go-dev/golang-gopls— package facts and call-site navigation.
