abstract

Documentation

Index

• type ArgsConsumer
• type Callbacker
• type CompleteCallbacker
• type Consumer
• type ErrorCallbacker
• type NextCallbacker
• type Observable
• type Observer
• type OnceSubscription
• type OperatorFunc
• type Productor
• type StringCallbacker
• type Subject
• type Subscription
  • func FreeSubs(subs []Subscription) (noClosedSubs []Subscription)

Types

type ArgsConsumer

type ArgsConsumer interface {
	Args() []interface{}
}

type Callbacker

type Callbacker interface{}

type CompleteCallbacker

type CompleteCallbacker interface {
	Complete()
}

type Consumer

type Consumer interface {
	Next(v interface{})
}

type ErrorCallbacker

type ErrorCallbacker interface {
	Err(error)
}

type NextCallbacker

type NextCallbacker interface {
	Next(interface{})
}

type Observable

type Observable interface {
	Subscribe(NextCallbacker) Subscription
	Pipe(...OperatorFunc) Observable
}

type Observer

type Observer interface {
	Next(interface{})
	Complete()
	Err(error)
}

type OnceSubscription

type OnceSubscription interface {
	Unsubscribe()
	Add(teardown interface{})
	// Done 某些生产者只能在程序结束了才会获得该channel
	// 调用该接口并获取到channel后将同时取消订阅
	Done() <-chan struct{}
	// Once 执行一次就获得channel,可确保获取到至少1次数据
	// 调用该接口并获取到channel后将同时取消订阅
	Once() <-chan struct{}
	// Try 执行过一次就获得channel,可确保获取到至少1次数据
	// 获取后不取消订阅
	Try() <-chan struct{}

	Remove(Subscription)
	IsClosed() bool
}

type OperatorFunc

type OperatorFunc func(interface{}) interface{}

type Productor

type Productor interface {
	Put(ctx context.Context, material interface{}) error
	// BoM() interface{}
	Run(ctx context.Context) error
	// Consumer() Consumer
	Consume(consumer Consumer) Subscription
}

type StringCallbacker

type StringCallbacker interface {
	Next(string)
}

type Subject

type Subject interface {
	Observable
	Next(v interface{})
}

type Subscription

type Subscription interface {
	OnceSubscription
	// Keeper 一直尝试，直到取消订阅或该订阅结束
	// 只有接收到调用接口后的数据才触发
	Keeper() <-chan struct{}
}

func FreeSubs

func FreeSubs(subs []Subscription) (noClosedSubs []Subscription)