PublishBehavior

README

PublishBehavior

PublishBehavior returns a Multicaster that shares a single subscription to the underlying Observable returning an initial value or the last value emitted by the underlying Observable. When the underlying Obervable terminates with an error, then subscribed observers will receive that error. After all observers have unsubscribed due to an error, the Multicaster does an internal reset just before the next observer subscribes.

Documentation

Overview

PublishBehavior returns a Multicaster that shares a single subscription to the underlying Observable returning an initial value or the last value emitted by the underlying Observable. When the underlying Obervable terminates with an error, then subscribed observers will receive that error. After all observers have unsubscribed due to an error, the Multicaster does an internal reset just before the next observer subscribes.

Example (PublishBehavior)

publish := From("b", "c").PublishBehavior("a")
go func() {
	time.Sleep(time.Microsecond)
	publish.Connect().Wait()
}()
publish.Println()
publish.Println()
publish.Println()
publish.Println()

Output:

a
b
c

Index

• Constants
• func MakeObserverObservable(age time.Duration, length int, capacity ...int) (Observer, Observable)
• type Connectable
  • func (c Connectable) Connect(schedulers ...Scheduler) Subscription
• type Multicaster
• type Observable
  • func Concat(observables ...Observable) Observable
  • func Empty() Observable
  • func From(slice ...interface{}) Observable
  • func Just(element interface{}) Observable
  • func Throw(err error) Observable
  • func Timer(initialDelay time.Duration, intervals ...time.Duration) Observable
  • func (o Observable) AsObservable() Observable
  • func (o Observable) AutoUnsubscribe() Observable
  • func (o Observable) ConcatWith(other ...Observable) Observable
  • func (o Observable) MapObservable(project func(interface{}) Observable) ObservableObservable
  • func (o Observable) MergeMapTo(inner Observable) Observable
  • func (o Observable) Multicast(factory func() Subject) Multicaster
  • func (o Observable) Println(a ...interface{}) error
  • func (o Observable) PublishBehavior(a interface{}) Multicaster
  • func (o Observable) Serialize() Observable
  • func (o Observable) Subscribe(observe Observer, schedulers ...Scheduler) Subscription
  • func (o Observable) Timeout(due time.Duration) Observable
• type ObservableObservable
  • func (o ObservableObservable) AutoUnsubscribe() ObservableObservable
  • func (o ObservableObservable) MergeAll() Observable
• type ObservableObserver
• type Observer
  • func (o Observer) AsObserver() Observer
  • func (o Observer) Complete()
  • func (o Observer) Error(err error)
  • func (o Observer) Next(next interface{})
• type RxError
  • func (e RxError) Error() string
• type Scheduler
  • func GoroutineScheduler() Scheduler
  • func NewScheduler() Scheduler
• type Subject
  • func NewBehaviorSubject(a interface{}) Subject
• type Subscriber
  • func NewSubscriber() Subscriber
• type Subscription

Examples

• Package (PublishBehavior)

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")

const TimeoutOccured = RxError("timeout occured")

TimeoutOccured is delivered to an observer if the stream times out.

Functions

func MakeObserverObservable

func MakeObserverObservable(age time.Duration, length int, capacity ...int) (Observer, Observable)

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 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.

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 Concat

func Concat(observables ...Observable) Observable

Concat emits the emissions from two or more Observables without interleaving them.

func Empty

func Empty() Observable

Empty creates an Observable that emits no items but terminates normally.

func From

func From(slice ...interface{}) Observable

From creates an Observable from multiple interface{} values passed in.

func Just

func Just(element interface{}) Observable

Just creates an Observable that emits a particular item.

func Throw

func Throw(err error) Observable

Throw creates an Observable that emits no items and terminates with an error.

func Timer

func Timer(initialDelay time.Duration, intervals ...time.Duration) Observable

Timer creates an Observable that emits a sequence of integers (starting at zero) after an initialDelay has passed. Subsequent values are emitted using a schedule of intervals passed in. If only the initialDelay is given, Timer will emit only once.

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) AutoUnsubscribe

func (o Observable) AutoUnsubscribe() Observable

AutoUnsubscribe will automatically unsubscribe from the source when it signals it is done. This Operator subscribes to the source Observable using a separate subscriber. When the source observable subsequently signals it is done, the separate subscriber will be Unsubscribed.

func (Observable) ConcatWith

func (o Observable) ConcatWith(other ...Observable) Observable

ConcatWith emits the emissions from two or more Observables without interleaving them.

func (Observable) MapObservable

func (o Observable) MapObservable(project func(interface{}) Observable) ObservableObservable

MapObservable transforms the items emitted by an Observable by applying a function to each item.

func (Observable) MergeMapTo

func (o Observable) MergeMapTo(inner Observable) Observable

MergeMapTo maps every entry emitted by the Observable into a single Observable. The stream of Observable items is then merged into a single stream of items using the MergeAll operator.

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) Println

func (o Observable) 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 (Observable) PublishBehavior

func (o Observable) PublishBehavior(a interface{}) Multicaster

PublishBehavior returns a Multicaster that shares a single subscription to the underlying Observable returning an initial value or the last value emitted by the underlying Observable. When the underlying Obervable terminates with an error, then subscribed observers will receive that error. After all observers have unsubscribed due to an error, the Multicaster does an internal reset just before the next observer subscribes.

func (Observable) Serialize

func (o Observable) Serialize() Observable

Serialize forces an Observable to make serialized calls and to be well-behaved.

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().

func (Observable) Timeout

func (o Observable) Timeout(due time.Duration) Observable

Timeout mirrors the source Observable, but issues an error notification if a particular period of time elapses without any emitted items. Timeout schedules a task on the scheduler passed to it during subscription.

type ObservableObservable

type ObservableObservable func(ObservableObserver, Scheduler, Subscriber)

ObservableObservable is a function taking an Observer, Scheduler and Subscriber. Calling it will subscribe the Observer to events from the Observable.

func (ObservableObservable) AutoUnsubscribe

func (o ObservableObservable) AutoUnsubscribe() ObservableObservable

AutoUnsubscribe will automatically unsubscribe from the source when it signals it is done. This Operator subscribes to the source Observable using a separate subscriber. When the source observable subsequently signals it is done, the separate subscriber will be Unsubscribed.

func (ObservableObservable) MergeAll

func (o ObservableObservable) MergeAll() Observable

MergeAll flattens a higher order observable by merging the observables it emits.

type ObservableObserver

type ObservableObserver func(next Observable, err error, done bool)

ObservableObserver 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 Observer

type Observer func(next interface{}, err error, done bool)

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

func (o Observer) AsObserver() Observer

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.

func (Observer) Error

func (o Observer) Error(err error)

Error is called by an Observable to report an error to the Observer.

func (Observer) Next

func (o Observer) Next(next interface{})

Next is called by an Observable to emit the next interface{} value to the Observer.

type RxError

type RxError string

func (RxError) Error

func (e RxError) Error() string

type Scheduler

type Scheduler = scheduler.Scheduler

Scheduler is used to schedule tasks to support subscribing and observing.

func GoroutineScheduler

func GoroutineScheduler() Scheduler

func NewScheduler

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 NewBehaviorSubject

func NewBehaviorSubject(a interface{}) Subject

NewBehaviorSubject returns a new BehaviorSubject. When an observer subscribes to a BehaviorSubject, it begins by emitting the item most recently emitted by the Observable part of the subject (or a seed/default value if none has yet been emitted) and then continues to emit any other items emitted later by 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

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.