README
¶
Retry
Retry if a source Observable sends an error notification, resubscribe to it in the hopes that it will complete without error. If count is zero or negative, the retry count will be effectively infinite. The scheduler passed when subscribing is used by Retry to schedule any retry attempt. The time between retries is 1 millisecond, so retry frequency is 1 kHz. Any SubscribeOn operators should be called after Retry to prevent lockups caused by mixing different schedulers in the same subscription for retrying and subscribing.
Documentation
¶
Overview ¶
Retry if a source Observable sends an error notification, resubscribe to it in the hopes that it will complete without error. If count is zero or negative, the retry count will be effectively infinite. The scheduler passed when subscribing is used by Retry to schedule any retry attempt. The time between retries is 1 millisecond, so retry frequency is 1 kHz. Any SubscribeOn operators should be called after Retry to prevent lockups caused by mixing different schedulers in the same subscription for retrying and subscribing.
Example (Retry) ¶
errored := false
a := CreateInt(func(N NextInt, E Error, C Complete, X Canceled) {
N(1)
N(2)
N(3)
if errored {
C()
} else {
// Error triggers subscribe and subscribe is scheduled on trampoline....
E(RxError("error"))
errored = true
}
})
err := a.Retry().Println()
fmt.Println(errored)
fmt.Println(err)
Output: 1 2 3 1 2 3 true <nil>
Example (RetryConcurrent) ¶
scheduler := GoroutineScheduler()
errored := false
a := CreateInt(func(N NextInt, E Error, C Complete, X Canceled) {
N(1)
N(2)
N(3)
if errored {
C()
} else {
E(RxError("error"))
errored = true
}
})
err := a.Retry().SubscribeOn(scheduler).Println()
fmt.Println(errored)
fmt.Println(err)
Output: 1 2 3 1 2 3 true <nil>
Index ¶
- Constants
- func MakeObserverObservable(age time.Duration, length int, capacity ...int) (Observer, Observable)
- type Canceled
- type Complete
- type Connectable
- type Error
- type IntObserver
- type Multicaster
- type NextInt
- type Observable
- type ObservableInt
- type Observer
- type RxError
- type Scheduler
- type Subject
- type Subscriber
- type Subscription
Examples ¶
Constants ¶
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")
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 Connectable ¶
type Connectable func(Scheduler, Subscriber)
Connectable provides the Connect method for a Multicaster.
func (Connectable) Connect ¶
func (c Connectable) Connect(schedulers ...Scheduler) Subscription
Connect instructs a multicaster to subscribe to its source and begin multicasting items to its subscribers. Connect accepts an optional scheduler argument.
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.
type Multicaster ¶
type Multicaster struct {
Observable
Connectable
}
Multicaster is a multicasting connectable observable. One or more Observers can subscribe to it simultaneously. It will subscribe to the source Observable when Connect is called. After that, every emission from the source is multcast to all subscribed Observers.
func (Multicaster) RefCount ¶
func (o Multicaster) RefCount() Observable
RefCount makes a Multicaster behave like an ordinary Observable. On first Subscribe it will call Connect on its Multicaster and when its last subscriber is Unsubscribed it will cancel the source connection by calling Unsubscribe on the subscription returned by the call to Connect.
type NextInt ¶
type NextInt func(int)
NextInt can be called to emit the next value to the IntObserver.
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 Defer ¶
func Defer(factory func() Observable) Observable
Defer does not create the Observable until the observer subscribes. It creates a fresh Observable for each subscribing observer. Use it to create observables that maintain separate state per subscription.
func From ¶
func From(slice ...interface{}) Observable
From creates an Observable from multiple interface{} values passed in.
func Throw ¶
func Throw(err error) Observable
Throw creates an Observable that emits no items and terminates with an error.
func (Observable) AsObservable ¶
func (o Observable) AsObservable() Observable
AsObservable returns the source Observable unchanged. This is a special case needed for internal plumbing.
func (Observable) Multicast ¶
func (o Observable) Multicast(factory func() Subject) Multicaster
Multicast converts an ordinary observable into a multicasting connectable observable or multicaster for short. A multicaster will only start emitting values after its Connect method has been called. The factory method passed in should return a new Subject that implements the actual multicasting behavior.
func (Observable) PublishLast ¶
func (o Observable) PublishLast() Multicaster
PublishLast returns a Multicaster that shares a single subscription to the underlying Observable containing only the last value emitted before it completes. When the underlying Obervable terminates with an error, then subscribed observers will receive only that error (and no value). After all observers have unsubscribed due to an error, the Multicaster does an internal reset just before the next observer subscribes.
func (Observable) Retry ¶
func (o Observable) Retry(count ...int) Observable
Retry if a source Observable sends an error notification, resubscribe to it in the hopes that it will complete without error. If count is zero or negative, the retry count will be effectively infinite. The scheduler passed when subscribing is used by Retry to schedule any retry attempt. The time between retries is 1 millisecond, so retry frequency is 1 kHz. Any SubscribeOn operators should be called after Retry to prevent lockups caused by mixing different schedulers in the same subscription for retrying and subscribing.
func (Observable) Subscribe ¶
func (o Observable) Subscribe(observe Observer, 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 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 CreateInt ¶
func CreateInt(create func(NextInt, Error, Complete, Canceled)) ObservableInt
CreateInt provides a way of creating an ObservableInt from scratch by calling observer methods programmatically.
The create function provided to CreateInt will be called once to implement the observable. It is provided with a NextInt, Error, Complete and Canceled function that can be called by the code that implements the Observable.
func (ObservableInt) Println ¶
func (o ObservableInt) Println(a ...interface{}) error
Println subscribes to the Observable and prints every item to os.Stdout while it waits for completion or error. Returns either the error or nil when the Observable completed normally. Println uses a serial scheduler created with NewScheduler().
func (ObservableInt) Retry ¶
func (o ObservableInt) Retry(count ...int) ObservableInt
Retry if a source ObservableInt sends an error notification, resubscribe to it in the hopes that it will complete without error. If count is zero or negative, the retry count will be effectively infinite. The scheduler passed when subscribing is used by Retry to schedule any retry attempt. The time between retries is 1 millisecond, so retry frequency is 1 kHz. Any SubscribeOn operators should be called after Retry to prevent lockups caused by mixing different schedulers in the same subscription for retrying and subscribing.
func (ObservableInt) SubscribeOn ¶
func (o ObservableInt) SubscribeOn(scheduler Scheduler) ObservableInt
SubscribeOn specifies the scheduler an ObservableInt 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) AsObserver ¶
AsObserver converts an observer of interface{} items to an observer of interface{} items.
func (Observer) Complete ¶
func (o Observer) Complete()
Complete is called by an Observable to signal that no more data is forthcoming to the Observer.
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 Subject ¶
type Subject struct {
Observer
Observable
}
Subject is a combination of an Observer and Observable. 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 Subject exposes all methods from the embedded Observer and Observable. Use the Observer Next, Error and Complete methods to feed data to it. Use the Observable 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 NewXxxSubject functions for more info.
func NewAsyncSubject ¶
func NewAsyncSubject() Subject
NewAsyncSubject creates a a subject that emits the last value (and only the last value) emitted by the Observable part, and only after that Observable part completes. (If the Observable part does not emit any values, the AsyncSubject also completes without emitting any values.)
It will also emit this same final value to any subsequent observers. However, if the Observable part terminates with an error, the AsyncSubject will not emit any items, but will simply pass along the error notification from the Observable part.
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