streaming

Documentation

Index

• type FlushReport
• type Msg
• type MsgProcessor
  • func AutoFlushed(processor MsgProcessor, ctx context.Context, interval time.Duration) MsgProcessor
  • func Buffered(dst MsgProcessor, capacity int) MsgProcessor
  • func HandleErrors(processor MsgProcessor, ch chan<- error) MsgProcessor
  • func HandleEvents(processor MsgProcessor, chErr chan<- error, chProcess chan<- ProcessReport, ...) MsgProcessor
  • func HandleFlush(processor MsgProcessor, ch chan<- FlushReport) MsgProcessor
  • func HandleProcess(processor MsgProcessor, ch chan<- ProcessReport) MsgProcessor
  • func Sharded(keyProvider ShardKeyProvider, processors ...MsgProcessor) MsgProcessor
  • func Transformed(processor MsgProcessor, transformer Transformer) MsgProcessor
• type ProcessReport
• type ShardKeyProvideFunc
  • func (s ShardKeyProvideFunc) ProvideShardKey(item interface{}) ([]byte, error)
• type ShardKeyProvider
• type Source
• type Stream
  • func New(src Source, dst MsgProcessor) Stream
• type TransformFunc
  • func (tf TransformFunc) Transform(original interface{}) (interface{}, error)
• type Transformer

Types

type FlushReport

type FlushReport interface {
	Duration() time.Duration
	Error() error
}

type Msg

type Msg interface {
	Value() interface{}
	Ack()
}

type MsgProcessor

type MsgProcessor interface {
	Process(messages []Msg) error
	Flush() error
}

func AutoFlushed

func AutoFlushed(processor MsgProcessor, ctx context.Context, interval time.Duration) MsgProcessor

func Buffered

func Buffered(dst MsgProcessor, capacity int) MsgProcessor

func HandleErrors

func HandleErrors(processor MsgProcessor, ch chan<- error) MsgProcessor

func HandleEvents

func HandleEvents(processor MsgProcessor, chErr chan<- error,
	chProcess chan<- ProcessReport, chFlush chan<- FlushReport) MsgProcessor

func HandleFlush

func HandleFlush(processor MsgProcessor, ch chan<- FlushReport) MsgProcessor

func HandleProcess

func HandleProcess(processor MsgProcessor, ch chan<- ProcessReport) MsgProcessor

func Sharded

func Sharded(keyProvider ShardKeyProvider, processors ...MsgProcessor) MsgProcessor

func Transformed

func Transformed(processor MsgProcessor, transformer Transformer) MsgProcessor

type ProcessReport

type ProcessReport interface {
	Amount() int
	Duration() time.Duration
	Error() error
}

type ShardKeyProvideFunc

type ShardKeyProvideFunc func(item interface{}) ([]byte, error)

func (ShardKeyProvideFunc) ProvideShardKey

func (s ShardKeyProvideFunc) ProvideShardKey(item interface{}) ([]byte, error)

type ShardKeyProvider

type ShardKeyProvider interface {
	ProvideShardKey(item interface{}) ([]byte, error)
}

type Source

type Source interface {
	Elements() <-chan Msg
}

type Stream

type Stream interface {
	Process(ctx context.Context)
}

func New

func New(src Source, dst MsgProcessor) Stream

type TransformFunc

type TransformFunc func(original interface{}) (interface{}, error)

func (TransformFunc) Transform

func (tf TransformFunc) Transform(original interface{}) (interface{}, error)

type Transformer

type Transformer interface {
	Transform(original interface{}) (interface{}, error)
}