README
¶
ReplaySubject
ReplaySubject ensures that all observers see the same sequence of emitted items, even if they subscribe after.
When bufferCapacity argument is 0, then DefaultReplayCapacity is used (currently 16383). When windowDuration argument is 0, then entries added to the buffer will remain fresh forever.
Documentation
¶
Overview ¶
ReplaySubject ensures that all observers see the same sequence of emitted items, even if they subscribe after.
When bufferCapacity argument is 0, then DefaultReplayCapacity is used (currently 16383). When windowDuration argument is 0, then entries added to the buffer will remain fresh forever.
Example (ReplaySubject) ¶
ReplaySubject is both an observer and an observable. The maximum buffersize is passed in along with how long entries in the replay buffer stay fresh. As long as entries are fresh they are send to an observer that subscribes.
subject := NewReplaySubjectInt(10, time.Hour)
// Feed the subject on a separate goroutine
go func() {
subject.Next(123)
subject.Next(456)
subject.Complete()
subject.Next(888)
}()
// Subscribe to subject on the default Trampoline scheduler.
subcription := subject.Subscribe(func(next int, err error, done bool) {
if !done {
fmt.Println("first", next)
}
})
subcription.Wait()
fmt.Println("--")
// Subscribe to subject again on the default Trampoline scheduler.
subcription = subject.Subscribe(func(next int, err error, done bool) {
if !done {
fmt.Println("second", next)
}
})
subcription.Wait()
Output: first 123 first 456 -- second 123 second 456
Example (ReplaySubjectMultiple) ¶
ReplaySubject example with multiple subscribers. Subscribe normally uses a synchronous Trampoline scheduler. To be able to subscribe multiple times without blocking, we have to change the scheduler to an asynchronous one using the SubscribeOn operator.
subject := NewReplaySubjectString(1000, time.Hour)
// Subscribe to subject on a goroutine
source := subject.SubscribeOn(GoroutineScheduler())
var results []*[]string
var subscriptions []Subscription
// Create a lot of subscribers running on separate goroutines.
for i := 0; i < 16; i++ {
result := make([]string, 0, 1000)
results = append(results, &result)
subscription := source.Subscribe(func(next string, err error, done bool) {
if !done {
result = append(result, next)
}
})
subscriptions = append(subscriptions, subscription)
}
// Feed the subject from the main goroutine
expect := make([]string, 0, 1000)
for i := 0; i < 1000; i++ {
s := fmt.Sprintf("Next %d", i)
expect = append(expect, s)
subject.Next(s)
}
subject.Complete()
// Wait for all subscriptions to complete
for _, subscription := range subscriptions {
subscription.Wait()
}
// Check the results for missing values
for index, result := range results {
identical := false
result := *result
if len(result) == len(expect) {
identical = true
for index, value := range result {
if value != expect[index] {
identical = false
}
}
}
if identical {
fmt.Printf("results %d as expected\n", index)
}
}
Output: results 0 as expected results 1 as expected results 2 as expected results 3 as expected results 4 as expected results 5 as expected results 6 as expected results 7 as expected results 8 as expected results 9 as expected results 10 as expected results 11 as expected results 12 as expected results 13 as expected results 14 as expected results 15 as expected
Index ¶
- Constants
- func MakeObserverObservable(age time.Duration, length int, capacity ...int) (Observer, Observable)
- type IntObserver
- type Observable
- type ObservableInt
- type ObservableString
- type Observer
- type RxError
- type Scheduler
- type StringObserver
- type SubjectInt
- type SubjectString
- type Subscriber
- type Subscription
Examples ¶
Constants ¶
const DefaultReplayCapacity = 16383
DefaultReplayCapacity is the default capacity of a replay buffer when a bufferCapacity of 0 is passed to the NewReplaySubject function.
const ErrUnsubscribed = RxError("subscriber unsubscribed")
Unsubscribed is the error returned by wait when the Unsubscribe method is called on the subscription.
const OutOfSubscriptions = RxError("out of subscriptions")
const TypecastFailed = RxError("typecast failed")
ErrTypecast is delivered to an observer if the generic value cannot be typecast to a specific type.
Variables ¶
This section is empty.
Functions ¶
func MakeObserverObservable ¶
MakeObserverObservable turns an observer into a multicasting and buffering observable. Both the observer and the obeservable are returned. These are then used as the core of any Subject implementation. The Observer side is used to pass items into the buffering multicaster. This then multicasts the items to every Observer that subscribes to the returned Observable.
age age below which items are kept to replay to a new subscriber. length length of the item buffer, number of items kept to replay to a new subscriber. [cap] Capacity of the item buffer, number of items that can be observed before blocking. [scap] Capacity of the subscription list, max number of simultaneous subscribers.
Types ¶
type IntObserver ¶
IntObserver is a function that gets called whenever the Observable has something to report. The next argument is the item value that is only valid when the done argument is false. When done is true and the err argument is not nil, then the Observable has terminated with an error. When done is true and the err argument is nil, then the Observable has completed normally.
func (IntObserver) Complete ¶
func (o IntObserver) Complete()
Complete is called by an ObservableInt to signal that no more data is forthcoming to the Observer.
func (IntObserver) Error ¶
func (o IntObserver) Error(err error)
Error is called by an ObservableInt to report an error to the Observer.
func (IntObserver) Next ¶
func (o IntObserver) Next(next int)
Next is called by an ObservableInt to emit the next int value to the Observer.
type Observable ¶
type Observable func(Observer, Scheduler, Subscriber)
Observable is a function taking an Observer, Scheduler and Subscriber. Calling it will subscribe the Observer to events from the Observable.
func (Observable) AsObservableInt ¶
func (o Observable) AsObservableInt() ObservableInt
AsObservableInt turns an Observable of interface{} into an ObservableInt. If during observing a typecast fails, the error ErrTypecastToInt will be emitted.
func (Observable) AsObservableString ¶
func (o Observable) AsObservableString() ObservableString
AsObservableString turns an Observable of interface{} into an ObservableString. If during observing a typecast fails, the error ErrTypecastToString will be emitted.
type ObservableInt ¶
type ObservableInt func(IntObserver, Scheduler, Subscriber)
ObservableInt is a function taking an Observer, Scheduler and Subscriber. Calling it will subscribe the Observer to events from the Observable.
func (ObservableInt) Subscribe ¶
func (o ObservableInt) Subscribe(observe IntObserver, schedulers ...Scheduler) Subscription
Subscribe operates upon the emissions and notifications from an Observable. This method returns a Subscription. Subscribe uses a serial scheduler created with NewScheduler().
type ObservableString ¶
type ObservableString func(StringObserver, Scheduler, Subscriber)
ObservableString is a function taking an Observer, Scheduler and Subscriber. Calling it will subscribe the Observer to events from the Observable.
func (ObservableString) Subscribe ¶
func (o ObservableString) Subscribe(observe StringObserver, schedulers ...Scheduler) Subscription
Subscribe operates upon the emissions and notifications from an Observable. This method returns a Subscription. Subscribe uses a serial scheduler created with NewScheduler().
func (ObservableString) SubscribeOn ¶
func (o ObservableString) SubscribeOn(scheduler Scheduler) ObservableString
SubscribeOn specifies the scheduler an ObservableString should use when it is subscribed to.
type Observer ¶
Observer is a function that gets called whenever the Observable has something to report. The next argument is the item value that is only valid when the done argument is false. When done is true and the err argument is not nil, then the Observable has terminated with an error. When done is true and the err argument is nil, then the Observable has completed normally.
func (Observer) AsIntObserver ¶
func (o Observer) AsIntObserver() IntObserver
AsIntObserver converts an observer of interface{} items to an observer of int items.
func (Observer) AsStringObserver ¶
func (o Observer) AsStringObserver() StringObserver
AsStringObserver converts an observer of interface{} items to an observer of string items.
type Scheduler ¶
Scheduler is used to schedule tasks to support subscribing and observing.
func GoroutineScheduler ¶
func GoroutineScheduler() Scheduler
func NewScheduler ¶ added in v0.0.1
func NewScheduler() Scheduler
type StringObserver ¶
StringObserver is a function that gets called whenever the Observable has something to report. The next argument is the item value that is only valid when the done argument is false. When done is true and the err argument is not nil, then the Observable has terminated with an error. When done is true and the err argument is nil, then the Observable has completed normally.
func (StringObserver) Complete ¶
func (o StringObserver) Complete()
Complete is called by an ObservableString to signal that no more data is forthcoming to the Observer.
func (StringObserver) Error ¶
func (o StringObserver) Error(err error)
Error is called by an ObservableString to report an error to the Observer.
func (StringObserver) Next ¶
func (o StringObserver) Next(next string)
Next is called by an ObservableString to emit the next string value to the Observer.
type SubjectInt ¶
type SubjectInt struct {
IntObserver
ObservableInt
}
SubjectInt is a combination of an IntObserver and ObservableInt. Subjects are special because they are the only reactive constructs that support multicasting. The items sent to it through its observer side are multicasted to multiple clients subscribed to its observable side.
The SubjectInt exposes all methods from the embedded IntObserver and ObservableInt. Use the IntObserver Next, Error and Complete methods to feed data to it. Use the ObservableInt methods to subscribe to it.
After a subject has been terminated by calling either Error or Complete, it goes into terminated state. All subsequent calls to its observer side will be silently ignored. All subsequent subscriptions to the observable side will be handled according to the specific behavior of the subject. There are different types of subjects, see the different NewXxxSubjectInt functions for more info.
func NewReplaySubjectInt ¶
func NewReplaySubjectInt(bufferCapacity int, windowDuration time.Duration) SubjectInt
NewReplaySubjectInt creates a new ReplaySubject. ReplaySubject ensures that all observers see the same sequence of emitted items, even if they subscribe after. When bufferCapacity argument is 0, then DefaultReplayCapacity is used (currently 16380). When windowDuration argument is 0, then entries added to the buffer will remain fresh forever.
type SubjectString ¶
type SubjectString struct {
StringObserver
ObservableString
}
SubjectString is a combination of an StringObserver and ObservableString. Subjects are special because they are the only reactive constructs that support multicasting. The items sent to it through its observer side are multicasted to multiple clients subscribed to its observable side.
The SubjectString exposes all methods from the embedded StringObserver and ObservableString. Use the StringObserver Next, Error and Complete methods to feed data to it. Use the ObservableString methods to subscribe to it.
After a subject has been terminated by calling either Error or Complete, it goes into terminated state. All subsequent calls to its observer side will be silently ignored. All subsequent subscriptions to the observable side will be handled according to the specific behavior of the subject. There are different types of subjects, see the different NewXxxSubjectString functions for more info.
func NewReplaySubjectString ¶
func NewReplaySubjectString(bufferCapacity int, windowDuration time.Duration) SubjectString
NewReplaySubjectString creates a new ReplaySubject. ReplaySubject ensures that all observers see the same sequence of emitted items, even if they subscribe after. When bufferCapacity argument is 0, then DefaultReplayCapacity is used (currently 16380). When windowDuration argument is 0, then entries added to the buffer will remain fresh forever.
type Subscriber ¶
type Subscriber interface {
// A Subscriber is also a Subscription.
Subscription
// Add will create and return a new child Subscriber setup in such a way that
// calling Unsubscribe on the parent will also call Unsubscribe on the child.
// Calling the Unsubscribe method on the child will NOT propagate to the
// parent!
Add() Subscriber
// OnUnsubscribe will add the given callback function to the subscriber.
// The callback will be called when either the Unsubscribe of the parent
// or of the subscriber itself is called. If the subscription was already
// canceled, then the callback function will just be called immediately.
OnUnsubscribe(callback func())
// OnWait will register a callback to call when subscription Wait is called.
OnWait(callback func())
// Done will set the error internally and then cancel the subscription by
// calling the Unsubscribe method. A nil value for error indicates success.
Done(err error)
// Error returns the error set by calling the Done(err) method. As long as
// the subscriber is still subscribed Error will return nil.
Error() error
}
Subscriber is a Subscription with management functionality.
func NewSubscriber ¶ added in v0.0.1
func NewSubscriber() Subscriber
New will create and return a new Subscriber.
type Subscription ¶
type Subscription interface {
// Subscribed returns true when the subscription is currently active.
Subscribed() bool
// Unsubscribe will do nothing if the subscription is not active. If the
// state is still active however, it will be changed to canceled.
// Subsequently, it will call Unsubscribe on all child subscriptions added
// through Add, along with all methods added through OnUnsubscribe. When the
// subscription is canceled by calling Unsubscribe a call to the Wait method
// will return the error ErrUnsubscribed.
Unsubscribe()
// Canceled returns true when the subscription state is canceled.
Canceled() bool
// Wait will by default block the calling goroutine and wait for the
// Unsubscribe method to be called on this subscription.
// However, when OnWait was called with a callback wait function it will
// call that instead. Calling Wait on a subscription that has already been
// canceled will return immediately. If the subscriber was canceled by
// calling Unsubscribe, then the error returned is ErrUnsubscribed.
// If the subscriber was terminated by calling Done, then the error
// returned here is the one passed to Done.
Wait() error
}
Subscription is an interface that allows code to monitor and control a subscription it received.
Source Files