fusion

Dec 25, 2020

Changes in this version

+ type Log func(_ map[string]interface{})

+ func LogFrom(ctx context.Context) Log

+ type ProcFn func(ctx context.Context, stream <-chan Msg) error

+ func (pf ProcFn) Run(ctx context.Context, stream <-chan Msg) error

type Runner

+ Log Log

Sep 17, 2020

Changes in this version

+ var Retry = errors.New("retry message")

+ type Fn struct

+ Func func(ctx context.Context, msg Msg) error

+ Workers int

+ func (fn *Fn) Run(ctx context.Context, stream <-chan Msg) error

type LineStream

+ func (rd *LineStream) Out(ctx context.Context) (<-chan Msg, error)

type Proc

+ Run func(ctx context.Context, stream <-chan Msg) error

+ type Runner struct

+ DrainTime time.Duration

+ Logger Logger

+ Proc Proc

+ Stream Stream

+ func (fu Runner) Run(ctx context.Context) error

type Stream

+ Out func(ctx context.Context) (<-chan Msg, error)

+ type StreamFn func(ctx context.Context) (*Msg, error)

+ func (sf StreamFn) Out(ctx context.Context) (<-chan Msg, error)

Jul 23, 2020

Changes in this version

+ var Fail = errors.New("fail message")

type Msg

+ Attribs map[string]string

+ Key []byte

+ Val []byte

type Options

+ OnFinish func(msg Msg, err error)

+ Proc Proc

Jul 22, 2020

Changes in this version

+ type Msg struct

+ Ack func(success bool, cause error)

+ Payloads []Payload

+ func (msg *Msg) Clone() Msg

Jul 22, 2020

Changes in this version

type LineStream

+ Buffer int

type Options

+ Stages []Proc

+ type Proc interface

+ Process func(ctx context.Context, msg Message) (*Message, error)

+ type ProcFunc func(ctx context.Context, msg Message) (*Message, error)

+ func (pf ProcFunc) Process(ctx context.Context, msg Message) (*Message, error)

Jul 22, 2020

Changes in this version

+ type Fusion struct

+ func (fu *Fusion) Run(ctx context.Context) error

type LineStream

+ func (rd *LineStream) ConsumeFrom(ctx context.Context) (<-chan Message, error)

+ func (rd *LineStream) Err() error

type Message

+ Ack func(success bool, cause error)

+ Payloads []Payload

+ func (msg *Message) Clone() Message

+ type NoOpProcessor struct

+ func (NoOpProcessor) Process(_ context.Context, _ Message) (*Message, error)

type Options

+ DrainWithin time.Duration

+ type Payload struct

+ Key []byte

+ Val []byte

+ func (p Payload) Clone() Payload

type Processor

+ Process func(ctx context.Context, msg Message) (*Message, error)

+ type ProcessorFunc func(ctx context.Context, msg Message) (*Message, error)

+ func (pf ProcessorFunc) Process(ctx context.Context, msg Message) (*Message, error)

+ type Source interface

+ ConsumeFrom func(ctx context.Context) (<-chan Message, error)

+ type SourceFunc func(ctx context.Context) (*Message, error)

+ func (sf SourceFunc) ConsumeFrom(ctx context.Context) (<-chan Message, error)

Jul 21, 2020

Changes in this version

+ type LineStream struct

+ From io.Reader

+ Offset int

+ Size int

+ func (rd *LineStream) Close() error

+ func (rd *LineStream) Read(ctx context.Context, readFn ReadFn) error

Jul 20, 2020

Changes in this version

+ var ErrNoMessage = errors.New("no message available")

+ var Failed = errors.New("fail message")

+ var Skip = errors.New("skip message")

+ type Actor struct

+ func New(opts Options) *Actor

+ func (actor *Actor) Run(ctx context.Context) error

+ type Backoff interface

+ RetryAfter func(msg Message) time.Duration

+ func ConstBackoff(interval time.Duration) Backoff

+ func ExpBackoff(base float64, initialTimeout, maxTimeout time.Duration) Backoff

+ type DelayQueue interface

+ Dequeue func(ctx context.Context, readFn ReadFn) error

+ Enqueue func(msg Message) error

+ type InMemQ struct

+ func (q *InMemQ) Dequeue(ctx context.Context, readFn ReadFn) error

+ func (q *InMemQ) Enqueue(msg Message) error

+ type Logger interface

+ Debugf func(msg string, args ...interface{})

+ Errorf func(msg string, args ...interface{})

+ Infof func(msg string, args ...interface{})

+ Warnf func(msg string, args ...interface{})

+ type Message struct

+ Attempts int

+ Key []byte

+ Time time.Time

+ Val []byte

+ type Options struct

+ Backoff Backoff

+ Logger Logger

+ MaxRetries int

+ OnFailure func(msg Message, err error)

+ PollInterval time.Duration

+ Processor Processor

+ Queue DelayQueue

+ Stream Stream

+ Workers int

+ type Processor func(ctx context.Context, msg Message) error

+ type ReadFn func(ctx context.Context, msg Message) error

+ type Stream interface

+ Read func(ctx context.Context, readFn ReadFn) error