batchstreamengine

Documentation

Index

• type BatchStreamEngine
  • func New(processor goduck.BatchProcessor, maxBatchSize int, ...) *BatchStreamEngine
  • func NewFromEndpoint(e endpoint.Endpoint, decoder goduck.EndpointBatchDecoder, maxBatchSize int, ...) *BatchStreamEngine
  • func (e *BatchStreamEngine) Run(ctx context.Context) error

Types

type BatchStreamEngine

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

BatchStreamEngine is an engine that processes a batch of messages from a stream, with the order preserved.

func New

func New(
	processor goduck.BatchProcessor,
	maxBatchSize int,
	maxBatchTimeout time.Duration,
	streams []goduck.Stream,
) *BatchStreamEngine

New creates a new BackStreamEngine.

func NewFromEndpoint

func NewFromEndpoint(
	e endpoint.Endpoint,
	decoder goduck.EndpointBatchDecoder,
	maxBatchSize int,
	maxBatchTimeout time.Duration,
	streams []goduck.Stream,
) *BatchStreamEngine

NewFromEndpoint creates a BatchProcessor from a go-kit endpoint

func (*BatchStreamEngine) Run

func (e *BatchStreamEngine) Run(ctx context.Context) error

Run starts processing the messages, until @ctx is closed