mono

Documentation

Overview

Example

gen := func(ctx context.Context, sink mono.Sink) {
	sink.Success("World")
}
mono.
	Create(gen).
	Map(func(i Any) (o Any, err error) {
		o = "Hello " + i.(string) + "!"
		return
	}).
	DoOnNext(func(v Any) error {
		fmt.Println(v)
		return nil
	}).
	Subscribe(context.Background())

Index

• func IsSubscribeAsync(m Mono) bool
• func NewProcessor(sc scheduler.Scheduler, hook ProcessorFinallyHook) (Mono, Sink, Disposable)
• type Any
• type Combinator
• type Disposable
• type FlatMapper
• type Mono
  • func Create(gen func(ctx context.Context, s Sink)) Mono
  • func CreateOneshot(gen func(ctx context.Context, s Sink)) Mono
  • func Delay(delay time.Duration) Mono
  • func Empty() Mono
  • func Error(e error) Mono
  • func ErrorOneshot(e error) Mono
  • func Just(v Any) Mono
  • func JustOneshot(v Any) Mono
  • func JustOrEmpty(v Any) Mono
  • func Zip(first Mono, second Mono, rest ...Mono) Mono
  • func ZipCombine(cmb Combinator, itemHandler func(item *reactor.Item), sources ...Mono) Mono
  • func ZipCombineOneshot(cmb Combinator, itemHandler func(*reactor.Item), sources ...Mono) Mono
  • func ZipOneshot(first Mono, second Mono, rest ...Mono) Mono
• type ProcessorFinallyHook
• type Sink

Examples

• Package

Functions

func IsSubscribeAsync

func IsSubscribeAsync(m Mono) bool

func NewProcessor

func NewProcessor(sc scheduler.Scheduler, hook ProcessorFinallyHook) (Mono, Sink, Disposable)

Types

type Any

type Any = reactor.Any

Alias

type Combinator

type Combinator func(values ...*reactor.Item) (reactor.Any, error)

type Disposable

type Disposable = reactor.Disposable

Alias

type FlatMapper

type FlatMapper func(value reactor.Any) Mono

type Mono

type Mono interface {
	reactor.Publisher
	// Filter tests the result and replay it if predicate returns true. Otherwise complete without value.
	Filter(reactor.Predicate) Mono
	// Map transforms the item emitted by this Mono by applying a synchronous function to it.
	Map(reactor.Transformer) Mono
	// FlatMap transforms the item emitted by this Mono asynchronously, returning the value emitted by another Mono.
	FlatMap(FlatMapper) Mono
	// SubscribeOn set the scheduler.Scheduler when reactor.Subscriber subscribes this Mono.
	SubscribeOn(scheduler.Scheduler) Mono
	// Block subscribes to this Mono and block indefinitely until a next signal is received.
	// Returns that value/error, or nil if the Mono completes empty.
	Block(context.Context) (Any, error)
	// DoOnNext adds a behavior triggered when the Mono emits a data successfully.
	DoOnNext(reactor.FnOnNext) Mono
	// DoOnComplete adds a behavior triggered when the Mono completes successfully (includes empty).
	DoOnComplete(reactor.FnOnComplete) Mono
	// DoOnSubscribe adds a behavior (side-effect) triggered when the Mono is done being subscribed, that is to say when a Subscription has been produced by the Publisher and passed to the Subscriber.OnSubscribe(Subscription).
	DoOnSubscribe(reactor.FnOnSubscribe) Mono
	// DoOnError adds a behavior triggered when the Mono completes with an error.
	DoOnError(reactor.FnOnError) Mono
	// DoOnCancel adds a behavior triggered when the Mono is cancelled.
	DoOnCancel(reactor.FnOnCancel) Mono
	// DoFinally adds a behavior triggering after the Mono terminates for any reason, including cancellation.
	DoFinally(reactor.FnOnFinally) Mono
	// DoOnDiscard description.
	DoOnDiscard(reactor.FnOnDiscard) Mono
	// SwitchIfEmpty fallbacks to an alternative Mono if this mono is completed without data.
	SwitchIfEmpty(alternative Mono) Mono
	// SwitchIfError fallbacks to an alternative Mono if this mono is completed without an error.
	SwitchIfError(alternativeFunc func(error) Mono) Mono
	// SwitchValueIfError fallbacks to an alternative value if this mono is completed without an error.
	SwitchValueIfError(v Any) Mono
	// DelayElement delays this Mono element (Subscriber.OnNext signal) by a given duration.
	DelayElement(delay time.Duration) Mono
	// Timeout propagates a Error in case no item arrives within the given Duration.
	Timeout(timeout time.Duration) Mono
	// ZipWith combines the result from this mono and another into a tuple.Tuple.
	ZipWith(other Mono) Mono
	// ZipCombineWith combines the result from this mono and another, you can customize the Combinator.
	ZipCombineWith(other Mono, cmb Combinator) Mono
	// Raw returns an internal RawPublisher.
	Raw() reactor.RawPublisher
}

Mono is a Reactive Streams Publisher with basic rx operators that completes successfully by emitting an element, or with an error.

func Create

func Create(gen func(ctx context.Context, s Sink)) Mono

func CreateOneshot

func CreateOneshot(gen func(ctx context.Context, s Sink)) Mono

func Delay

func Delay(delay time.Duration) Mono

func Empty

func Empty() Mono

func Error

func Error(e error) Mono

func ErrorOneshot

func ErrorOneshot(e error) Mono

func Just

func Just(v Any) Mono

func JustOneshot

func JustOneshot(v Any) Mono

func JustOrEmpty

func JustOrEmpty(v Any) Mono

func Zip

func Zip(first Mono, second Mono, rest ...Mono) Mono

func ZipCombine

func ZipCombine(cmb Combinator, itemHandler func(item *reactor.Item), sources ...Mono) Mono

func ZipCombineOneshot

func ZipCombineOneshot(cmb Combinator, itemHandler func(*reactor.Item), sources ...Mono) Mono

func ZipOneshot

func ZipOneshot(first Mono, second Mono, rest ...Mono) Mono

type ProcessorFinallyHook

type ProcessorFinallyHook func(reactor.SignalType, reactor.Disposable)

type Sink

type Sink interface {
	Success(Any)
	Error(error)
}