rx

Documentation

Index

• Constants
• func IsWrongTupleTypeError(err error) bool
• func NewSubscriberFacade(s Subscriber) reactor.Subscriber
• type FnFinally
• type FnOnCancel
• type FnOnComplete
• type FnOnError
• type FnOnNext
• type FnOnRequest
• type FnOnSubscribe
• type FnPredicate
• type FnTransform
• type Item
• type Publisher
• type RawPublisher
• type SignalType
  • func (s SignalType) String() string
• type Subscriber
  • func NewSubscriber(opts ...SubscriberOption) Subscriber
• type SubscriberOption
  • func OnComplete(onComplete FnOnComplete) SubscriberOption
  • func OnError(onError FnOnError) SubscriberOption
  • func OnNext(onNext FnOnNext) SubscriberOption
  • func OnSubscribe(onSubscribe FnOnSubscribe) SubscriberOption
• type Subscription
• type Tuple
  • func NewTuple(t ...*reactor.Item) Tuple
  • func (t Tuple) CollectValues() (values []payload.Payload)
  • func (t Tuple) First() (payload.Payload, error)
  • func (t Tuple) ForEach(callback func(payload.Payload, error) bool)
  • func (t Tuple) ForEachWithIndex(callback func(payload.Payload, error, int) bool)
  • func (t Tuple) Get(index int) (payload.Payload, error)
  • func (t Tuple) GetValue(index int) payload.Payload
  • func (t Tuple) HasError() bool
  • func (t Tuple) Last() (payload.Payload, error)
  • func (t Tuple) Len() int
  • func (t Tuple) Second() (payload.Payload, error)

Constants

const (
	// SignalComplete indicated that subscriber was completed.
	SignalComplete = SignalType(reactor.SignalTypeComplete)
	// SignalCancel indicates that subscriber was cancelled.
	SignalCancel = SignalType(reactor.SignalTypeCancel)
	// SignalError indicates that subscriber has some faults.
	SignalError = SignalType(reactor.SignalTypeError)
)

const RequestMax = reactor.RequestInfinite

RequestMax represents unbounded request amount.

Functions

func IsWrongTupleTypeError

func IsWrongTupleTypeError(err error) bool

IsWrongTupleTypeError returns true if target error is type of wrong tuple type.

func NewSubscriberFacade

func NewSubscriberFacade(s Subscriber) reactor.Subscriber

Types

type FnFinally

type FnFinally = func(s SignalType)

FnFinally is alias of function for signal when all things done.

type FnOnCancel

type FnOnCancel = func()

FnOnCancel is alias of function for signal when subscription canceled.

type FnOnComplete

type FnOnComplete = func()

FnOnComplete is alias of function for signal when no more elements are available

type FnOnError

type FnOnError = func(e error)

FnOnError is alias of function for signal when an error occurred.

type FnOnNext

type FnOnNext = func(input payload.Payload) error

FnOnNext is alias of function for signal when next element arrived.

type FnOnRequest

type FnOnRequest = func(n int)

FnOnRequest is alias of function for signal when requesting next element.

type FnOnSubscribe

type FnOnSubscribe = func(ctx context.Context, s Subscription)

FnOnSubscribe is alias of function for signal when subscribe begin.

type FnPredicate

type FnPredicate = func(input payload.Payload) bool

FnPredicate is alias of function for filter operations.

type FnTransform

type FnTransform = func(payload.Payload) (payload.Payload, error)

FnTransform is alias of function to transform a payload to another.

type Item

type Item struct {
	V payload.Payload
	E error
}

Item is a kind of container which contains value or error.

type Publisher

type Publisher interface {
	RawPublisher
	// Subscribe subscribe elements from a publisher, returns a Disposable.
	// You can add some custom options.
	// Using `OnSubscribe`, `OnNext`, `OnComplete` and `OnError` as handler wrapper.
	Subscribe(ctx context.Context, options ...SubscriberOption)
}

Publisher is a provider of a potentially unbounded number of sequenced elements, \ publishing them according to the demand received from its Subscriber(s).

type RawPublisher

type RawPublisher interface {
	// SubscribeWith can be used to subscribe current publisher.
	SubscribeWith(ctx context.Context, s Subscriber)
}

RawPublisher represents a basic Publisher which can be subscribed by a Subscriber.

type SignalType

type SignalType reactor.SignalType

SignalType is the signal of reactive events like `OnNext`, `OnComplete`, `OnCancel` and `OnError`.

func (SignalType) String

func (s SignalType) String() string

type Subscriber

type Subscriber interface {
	// OnNext represents data notification sent by the Publisher in response to requests to Subscription#Request.
	OnNext(payload payload.Payload)
	// OnError represents failed terminal state.
	OnError(error)
	// OnComplete represents successful terminal state.
	OnComplete()
	// OnSubscribe invoked after Publisher subscribed.
	// No data will start flowing until Subscription#Request is invoked.
	OnSubscribe(context.Context, Subscription)
}

Subscriber will receive call to OnSubscribe(Subscription) once after passing an instance of Subscriber to Publisher#SubscribeWith

var (
	// EmptySubscriber is a blank Subscriber.
	EmptySubscriber Subscriber = &subscriber{}
	// EmptyRawSubscriber is a blank native Subscriber in reactor-go.
	EmptyRawSubscriber = reactor.NewSubscriber(reactor.OnNext(func(v reactor.Any) error {
		return nil
	}))
)

func NewSubscriber

func NewSubscriber(opts ...SubscriberOption) Subscriber

NewSubscriber create a new Subscriber with custom options.

type SubscriberOption

type SubscriberOption func(*subscriber)

SubscriberOption is option of subscriber. You can call OnNext, OnComplete, OnError or OnSubscribe.

func OnComplete

func OnComplete(onComplete FnOnComplete) SubscriberOption

OnComplete returns s SubscriberOption handling Complete event.

func OnError

func OnError(onError FnOnError) SubscriberOption

OnError returns s SubscriberOption handling Error event.

func OnNext

func OnNext(onNext FnOnNext) SubscriberOption

OnNext returns s SubscriberOption handling Next event.

func OnSubscribe

func OnSubscribe(onSubscribe FnOnSubscribe) SubscriberOption

OnSubscribe returns s SubscriberOption handling Subscribe event.

type Subscription

type Subscription = reactor.Subscription

Subscription represents a one-to-one lifecycle of a Subscriber subscribing to a Publisher.

type Tuple

type Tuple []*reactor.Item

Tuple is a container contains multiple items.

func NewTuple

func NewTuple(t ...*reactor.Item) Tuple

NewTuple returns a new Tuple.

func (Tuple) CollectValues

func (t Tuple) CollectValues() (values []payload.Payload)

CollectValues collects values and returns a slice.

func (Tuple) First

func (t Tuple) First() (payload.Payload, error)

First returns the first value or error.

func (Tuple) ForEach

func (t Tuple) ForEach(callback func(payload.Payload, error) bool)

ForEach visits each item in the Tuple.

func (Tuple) ForEachWithIndex

func (t Tuple) ForEachWithIndex(callback func(payload.Payload, error, int) bool)

ForEachWithIndex visits each item in the Tuple with index.

func (Tuple) Get

func (t Tuple) Get(index int) (payload.Payload, error)

Get returns the value or error with custom index.

func (Tuple) GetValue

func (t Tuple) GetValue(index int) payload.Payload

GetValue returns the value with custom index.

func (Tuple) HasError

func (t Tuple) HasError() bool

HasError returns true if this Tuple contains error.

func (Tuple) Last

func (t Tuple) Last() (payload.Payload, error)

Last returns the last value or error.

func (Tuple) Len

func (t Tuple) Len() int

Len returns the length of Tuple.

func (Tuple) Second

func (t Tuple) Second() (payload.Payload, error)

Second returns the second value or error.