teamai-test/.teamai/skills/common/golang-samber-ro/references/subjects-guide.md

Subjects Guide

Subjects are both Observable and Observer — they can receive values (via Send, Error, Complete) and be subscribed to. Subjects are natively hot: subscribers share a single execution, and late subscribers only see future emissions (unless replay is configured).

When to Use Subjects vs Cold Observables

Use case	Approach
Data pipeline from a known source (slice, channel, HTTP)	Cold observable (default)
Event bus where producers and consumers are decoupled	Subject
Multiple consumers need the same WebSocket/ticker stream	Cold observable + Share() or ShareReplay()
Imperatively push values from non-reactive code	Subject
Bridge between callback API and reactive pipeline	Subject (receive callbacks, emit to pipeline)

Subject Types

PublishSubject

Standard multicast. Subscribers only see values emitted after they subscribe.

subject := ro.NewPublishSubject[string]()

// Subscriber 1
subject.Subscribe(ro.OnNext(func(s string) {
    fmt.Println("sub1:", s)
}))

subject.Send("hello")  // sub1 sees this

// Subscriber 2 (late)
subject.Subscribe(ro.OnNext(func(s string) {
    fmt.Println("sub2:", s)
}))

subject.Send("world")  // both see this
subject.Complete()

Use when: broadcasting events where late subscribers don't need history — UI events, log streams, notifications.

BehaviorSubject

Replays the last emitted value (or the initial value) to every new subscriber immediately on subscription.

subject := ro.NewBehaviorSubject[int](0) // initial value = 0

// Subscriber 1 immediately receives 0
subject.Subscribe(ro.OnNext(func(v int) {
    fmt.Println("sub1:", v) // 0, then 42
}))

subject.Send(42)

// Subscriber 2 immediately receives 42 (latest value)
subject.Subscribe(ro.OnNext(func(v int) {
    fmt.Println("sub2:", v) // 42
}))

Use when: subscribers need the current state — config values, connection status, latest price.

ReplaySubject

Buffers the last N values and replays them to every new subscriber.

subject := ro.NewReplaySubject[string](3) // buffer size = 3

subject.Send("a")
subject.Send("b")
subject.Send("c")
subject.Send("d") // "a" evicted from buffer

// Late subscriber receives "b", "c", "d" (last 3)
subject.Subscribe(ro.OnNext(func(s string) {
    fmt.Println(s)
}))

Use when: late subscribers need recent history — chat messages, recent logs, last N stock ticks.

AsyncSubject

Emits only the last value and only when the subject completes. If the subject errors, no value is emitted.

subject := ro.NewAsyncSubject[int]()

subject.Subscribe(ro.NewObserver(
    func(v int) { fmt.Println(v) },    // receives 3 only
    func(err error) { },
    func() { fmt.Println("done") },
))

subject.Send(1)
subject.Send(2)
subject.Send(3)
subject.Complete() // triggers emission of 3, then "done"

Use when: only the final result matters — computation result, last response in a batch.

UnicastSubject

Allows exactly one subscriber. Buffers values internally until that subscriber connects.

subject := ro.NewUnicastSubject[int](100) // buffer size

subject.Send(1) // buffered
subject.Send(2) // buffered

// Single subscriber receives buffered + future values
subject.Subscribe(ro.OnNext(func(v int) {
    fmt.Println(v) // 1, 2, then future values
}))
// Second subscribe would panic or error

Use when: single consumer with buffering — job queues, request pipelines where exactly one handler processes events.

Cold to Hot Conversion

When you have a cold observable (e.g. an HTTP request) but need multiple subscribers to share it:

Share

// Each subscriber to `cold` would trigger a separate HTTP request
cold := httpPlugin.Get[Data](url)

// Share: single execution, multiple subscribers
hot := ro.Pipe1(cold, ro.Share[Data]())

hot.Subscribe(uiObserver)       // shares one HTTP call
hot.Subscribe(metricsObserver)  // same data, no extra request

Share uses reference counting: the source subscribes when the first subscriber arrives and unsubscribes when the last one leaves.

ShareReplay

// Late subscribers get the last N values + future values
hot := ro.Pipe1(cold, ro.ShareReplay[Data](1))

Connectable Observable

For precise control over when the shared subscription starts:

connectable := ro.Connectable[Data](cold)

// Set up subscribers first
connectable.Subscribe(observer1)
connectable.Subscribe(observer2)

// Start the shared execution explicitly
sub, err := connectable.Connect(ctx)

Subject Decision Table

Subject	Replay	Subscribers	Use case
PublishSubject	None	Many	Event bus, notifications
BehaviorSubject	Last 1 (+ initial)	Many	Current state, config
ReplaySubject	Last N	Many	Recent history, chat
AsyncSubject	Last 1 (on complete)	Many	Final computation result
UnicastSubject	Buffered (pre-subscribe)	Exactly 1	Single-consumer queue

Common Subject Mistakes

Mistake	Why	Fix
Calling Send() after Complete()	Values are silently dropped — the subject is terminal	Track lifecycle, don't reuse completed subjects
Using PublishSubject when late subscribers need history	Late subscribers miss all prior events	Use BehaviorSubject (last 1) or ReplaySubject (last N)
Using ReplaySubject with unbounded buffer	Memory grows without limit	Set an explicit bufferSize
Multiple subscribers on UnicastSubject	Panics or undefined behavior	Use PublishSubject for multicast, UnicastSubject for single consumer
Not calling Complete() on subjects	Subscribers wait forever, goroutine leak	Always Complete() or Error() when the source is done