flux

Documentation

Overview

Example

gen := func(ctx context.Context, sink flux.Sink) {
	for i := 0; i < 10; i++ {
		v := i
		sink.Next(v)
	}
	sink.Complete()
}
done := make(chan struct{})

var su reactor.Subscription
flux.Create(gen).
	Filter(func(i Any) bool {
		return i.(int)%2 == 0
	}).
	Map(func(i interface{}) (Any, error) {
		return fmt.Sprintf("#HELLO_%04d", i.(int)), nil
	}).
	SubscribeOn(scheduler.Elastic()).
	Subscribe(context.Background(),
		reactor.OnSubscribe(func(ctx context.Context, s reactor.Subscription) {
			su = s
			s.Request(1)
		}),
		reactor.OnNext(func(v Any) error {
			fmt.Println("next:", v)
			su.Request(1)
			return nil
		}),
		reactor.OnComplete(func() {
			close(done)
		}),
	)
<-done

Index

• Constants
• func InitBuffSize(size int)
• type Any
• type CreateOption
  • func WithOverflowStrategy(o OverflowStrategy) CreateOption
• type Flux
  • func Create(c func(ctx context.Context, sink Sink), options ...CreateOption) Flux
  • func Empty() Flux
  • func Error(e error) Flux
  • func Interval(period time.Duration) Flux
  • func Just(values ...Any) Flux
  • func Range(start, count int) Flux
• type FnSwitchOnFirst
• type OverflowStrategy
• type Processor
  • func NewUnicastProcessor() Processor
• type Signal
• type Sink

Examples

• Package

Constants

const (
	BuffSizeXS = 32
	BuffSizeSM = 256
)

Functions

func InitBuffSize

func InitBuffSize(size int)

InitBuffSize initialize the size of buff. (default=256)

Types

type Any

type Any = reactor.Any

type CreateOption

type CreateOption func(*fluxCreate)

func WithOverflowStrategy

func WithOverflowStrategy(o OverflowStrategy) CreateOption

type Flux

type Flux interface {
	reactor.Publisher
	Filter(reactor.Predicate) Flux
	Map(reactor.Transformer) Flux
	Take(n int) Flux
	DoOnDiscard(reactor.FnOnDiscard) Flux
	DoOnNext(reactor.FnOnNext) Flux
	DoOnComplete(reactor.FnOnComplete) Flux
	DoOnError(reactor.FnOnError) Flux
	DoOnCancel(reactor.FnOnCancel) Flux
	DoOnRequest(reactor.FnOnRequest) Flux
	DoOnSubscribe(reactor.FnOnSubscribe) Flux
	DoFinally(reactor.FnOnFinally) Flux
	SwitchOnFirst(FnSwitchOnFirst) Flux
	DelayElement(delay time.Duration) Flux
	SubscribeOn(scheduler.Scheduler) Flux
	SubscribeWithChan(ctx context.Context, valueChan interface{}, errChan chan<- error)
	BlockFirst(context.Context) (Any, error)
	BlockLast(context.Context) (Any, error)
	BlockToSlice(ctx context.Context, slicePtr interface{}) error
}

func Create

func Create(c func(ctx context.Context, sink Sink), options ...CreateOption) Flux

func Empty

func Empty() Flux

func Error

func Error(e error) Flux

func Interval

func Interval(period time.Duration) Flux

func Just

func Just(values ...Any) Flux

func Range

func Range(start, count int) Flux

type FnSwitchOnFirst

type FnSwitchOnFirst func(s Signal, f Flux) Flux

type OverflowStrategy

type OverflowStrategy int8

const (
	OverflowBuffer OverflowStrategy = iota
	OverflowIgnore
	OverflowError
	OverflowDrop
	OverflowLatest
)

type Processor

type Processor interface {
	Flux
	Sink
}

func NewUnicastProcessor

func NewUnicastProcessor() Processor

type Signal

type Signal interface {
	Value() (Any, bool)
	Type() reactor.SignalType
}

type Sink

type Sink interface {
	Complete()
	Error(error)
	Next(Any)
}