ro

A Subject is a special type that acts as both an Observable and an Observer. It serves as a bridge or proxy that can multicast values to multiple observers, making it the primary tool for implementing 'hot' observables and event broadcasting patterns.

Key characteristics:

• Observable: It can be subscribed to like any other observable.
• Observer: It can receive values via Next, Error, and Complete methods.
• Multicaster: It broadcasts the same values to all its subscribers.
• Hot by nature: Values are shared among all subscribers concurrently, unlike 'cold' observables where each subscription gets its own independent stream of data.

type Subject[T any] interface {
    Observable[T]
    Observer[T]

    HasObserver() bool
    CountObservers() int
}

An Observer is the consumer side of reactive programming in samber/ro. It is the destination for values emitted by an Observable.

Key characteristics:

• Consumer of values: Receives values emitted by Observables.
• Notification handler: Processes Next, Error, and Complete notifications.
• Stateful: Tracks whether it is active, completed, or errored.
• Thread-safe: Multiple goroutines can safely call Observer methods.

An Observer follows a specific lifecycle: it starts as Active, and once it receives either an Error or a Complete notification, it transitions to a terminal state where no further notifications are accepted.

ro is a reactive programming library for Go that implements Observable streams. It is inspired by ReactiveX patterns and is designed to handle event-driven logic by treating events as streams that can be observed, transformed, and composed.

Key capabilities include:

• Handling asynchronous events naturally.
• Transforming and composing data streams declaratively.
• Managing backpressure and resource usage.
• Building responsive and resilient applications.

It is conceptually similar to samber/lo, but focused on events rather than collections.

An Observable is the core abstraction in ro representing a push-based stream of values over time. It acts as both a data producer and a factory for streams.

Key Characteristics:

• Producer of values: Emits zero or more values.
• Stream factory: Each subscription triggers a new, independent execution (for cold observables).
• Lazy: Values are only produced when a subscription occurs.
• Push-based: The producer pushes values to observers rather than the consumer pulling them.

Notification Types:

1. Next: Emits a value from the sequence.
2. Error: Emits an error and terminates the stream.
3. Complete: Signals successful completion and terminates the stream.

Note: Once an Error or Complete notification is emitted, no further values will be produced.

A Subscription manages the execution of an Observable. It acts as a resource manager for cleanup, a cancellation token to stop operations, and a lifecycle controller. It is thread-safe and can be used across multiple goroutines.

To prevent resource leaks, always capture the Subscription returned by .Subscribe() and call .Unsubscribe() when the operation is no longer needed.

// Create an Observable
observable := ro.Interval(1 * time.Second)

// Subscribe and get the subscription
subscription := observable.Subscribe(ro.OnNext(func(tick int64) {
    fmt.Println("Tick:", tick)
}))

// Cancel the subscription to prevent leaks
subscription.Unsubscribe()