pipeline

Jul 30, 2024

Changes in this version

type Option

+ func WithBatchProcessor(p goduck.BatchProcessor) Option

+ func WithProcessor(p goduck.Processor) Option

Dec 13, 2023

Changes in this version

+ var ErrInfiniteBehavior = errors.New("DLQ is active but max retries is set as infinite")

type Config

+ Backoffmiddleware struct{ ... }

Jun 15, 2023

Changes in this version

+ const EngineTypeRunOnce

type Config

+ EngineType string

Aug 16, 2022

Changes in this version

+ var ErrBothInputSet = errors.New("input streams and message pool cannot be set simultaneously")

+ var ErrEmptyInputStreamOrMessagePool = errors.New("input streams and message pool are nil or empty")

type Config

+ MessagePool MessagePoolConfig

+ type MessagePoolConfig struct

+ NWorkers int

+ Provider string

+ Pubsub pubsubqueue.PubsubConfigs

Nov 5, 2021

Changes in this version

+ const ShutdownPriorityEngine

+ const ShutdownPriorityStream

+ var ErrBatchSizeInvalid = errors.New("invalid batch size")

+ var ErrBothDecodersSet = errors.New("decoder and batch decoder cannot be set simultaneously")

+ var ErrEndpointNil = errors.New("endpoint is nil")

+ var ErrInputStreamNil = errors.New("input streams are nil or empty")

+ var ErrNilDecoders = errors.New("decoder and batch decoder are both nil")

+ var ErrShutdown = errors.New("shutdown didn't finished as expected")

+ var ErrSinkEncoderNil = errors.New("sink encoder is nil")

+ var ErrSinkNil = errors.New("sink is nil")

+ var ErrSystemNameEmpty = errors.New("system name is empty")

+ type Config struct

+ InputStream struct{ ... }

+ Kafka struct{ ... }

+ StaleAfter time.Duration

+ SystemName string

+ type Option func(*pipelineBuilderOptions)

+ func WithBatchDecoder(d goduck.EndpointBatchDecoder) Option

+ func WithConfig(userConfig Config) Option

+ func WithDecoder(d goduck.EndpointDecoder) Option

+ func WithEndpoint(e endpoint.Endpoint) Option

+ func WithInputStreams(s ...goduck.Stream) Option

+ func WithMiddlewares(middlewares ...endpoint.Middleware) Option

+ func WithSink(s Sink, e SinkEncoder) Option

+ type Pipeline interface

+ Run func(ctx context.Context) error

+ Shutdown func(ctx context.Context) error

+ func MustNew(opts ...Option) Pipeline

+ func New(opts ...Option) (Pipeline, error)

+ type Sink interface

+ Store func(ctx context.Context, input ...SinkMessage) error

+ func SinkWithRetry(next Sink) Sink

+ type SinkEncoder func(ctx context.Context, response interface{}) ([]SinkMessage, error)

+ type SinkMessage interface