luigi

type Broadcast ¶

type Broadcast interface {
	// Register a Sink for updates to be sent.
	Register(dst Sink) func()
}

Broadcast is an interface for registering one or more Sinks to recieve updates.

sink, bcast := NewBroadcast()
defer sink.Close()

var printOutput FuncSink = func(
	_ context.Context,
	v interface{},
	_ error,
) error {
	if v == nil {
		return nil
	}
	fmt.Println(*v.(*string))
	return nil
}

closeSink := bcast.Register(printOutput)
defer closeSink()
closeSink = bcast.Register(printOutput)
defer closeSink()

msg := "I sink this should be printed twice"
_ = sink.Pour(context.Background(), &msg)

Output:

I sink this should be printed twice
I sink this should be printed twice

type EOS ¶

type EOS struct{}

EOS stands for End Of Stream. It signals when a non-blocking stream is empty, or a stream is closed.

Similar the io package's EOF.

func (_ EOS) Error() string

type ErrorCloser ¶

type ErrorCloser interface {
	CloseWithError(error) error
}

type FuncSink ¶

type FuncSink func(ctx context.Context, v interface{}, err error) error

FuncSink defines a function which can be used as a Sink.

func (fSink FuncSink) Close() error

func (fSink FuncSink) CloseWithError(err error) error

func (fSink FuncSink) Pour(ctx context.Context, v interface{}) error

type FuncSource ¶

type FuncSource func(context.Context) (interface{}, error)

FuncSource defines a function which can be used as a Source.

func (fSink FuncSource) Next(ctx context.Context) (interface{}, error)

type Observable ¶

type Observable interface {
	// Broadcast allows subscribing to changes
	Broadcast

	// Set sets a new value
	Set(interface{}) error

	// Value returns the current value
	Value() (interface{}, error)
}

Observabe wraps an interface{} value and allows tracking changes to it

func NewObservable(v interface{}) Observable

type PipeOpt ¶

type PipeOpt func(*pipeOpts) error

PipeOpt configures NewPipes behavior

func NonBlocking() PipeOpt

func WithBuffer(bufSize int) PipeOpt

type PushSource ¶

type PushSource interface {
	Push(ctx context.Context, dst Sink) error
}

PushSource is the interface for requesting all content be written to the given sink.

type Sink ¶

type Sink interface {
	Pour(ctx context.Context, v interface{}) error
	Close() error
}

Sink is the interface which wraps methods writing to a stream.

type SliceSink ¶

type SliceSink struct {
	// contains filtered or unexported fields
}

SliceSink binds Sink methods to an interface array.

func NewSliceSink(arg *[]interface{}) *SliceSink

func (sink *SliceSink) Close() error

func (sink *SliceSink) Pour(ctx context.Context, v interface{}) error

type SliceSource ¶

type SliceSource []interface{}

SliceSink binds Source methods to an interface array.

func (src *SliceSource) Next(context.Context) (v interface{}, err error)

type Source ¶

type Source interface {
	Next(context.Context) (obj interface{}, err error)
}

Source is the interface which wraps the Next method for reading from a stream.