batcher

Documentation

Overview

Package batcher supports batching of items. Create a Batcher with a handler and add items to it. Items are accumulated while handler calls are in progress; when the handler returns, it will be called again with items accumulated since the last call. Multiple concurrent calls to the handler are supported.

Index

• Variables
• func Split(n int, opts *Options) []int
• type Batcher
  • func New(itemType reflect.Type, opts *Options, handler func(interface{}) error) *Batcher
  • func (b *Batcher) Add(ctx context.Context, item interface{}) error
  • func (b *Batcher) AddNoWait(item interface{}) <-chan error
  • func (b *Batcher) Shutdown()
• type Options
  • func (o *Options) NewMergedOptions(opts *Options) *Options

Variables

var ErrMessageTooLarge = errors.New("batcher: message too large")

Message is larger than the maximum batch byte size

Functions

func Split

func Split(n int, opts *Options) []int

Split determines how to split n (representing n items) into batches based on opts. It returns a slice of batch sizes.

For example, Split(10) might return [10], [5, 5], or [2, 2, 2, 2, 2] depending on opts. opts may be nil to accept defaults.

Split will return nil if n is less than o.MinBatchSize.

The sum of returned batches may be less than n (e.g., if n is 10x larger than o.MaxBatchSize, but o.MaxHandlers is less than 10).

Types

type Batcher

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

A Batcher batches items.

func New

func New(itemType reflect.Type, opts *Options, handler func(interface{}) error) *Batcher

New creates a new Batcher.

itemType is type that will be batched. For example, if you want to create batches of *Entry, pass reflect.TypeOf(&Entry{}) for itemType.

opts can be nil to accept defaults.

handler is a function that will be called on each bundle. If itemExample is of type T, the argument to handler is of type []T.

func (*Batcher) Add

func (b *Batcher) Add(ctx context.Context, item interface{}) error

Add adds an item to the batcher. It blocks until the handler has processed the item and reports the error that the handler returned. If Shutdown has been called, Add immediately returns an error.

func (*Batcher) AddNoWait

func (b *Batcher) AddNoWait(item interface{}) <-chan error

AddNoWait adds an item to the batcher and returns immediately. When the handler is called on the item, the handler's error return value will be sent to the channel returned from AddNoWait.

func (*Batcher) Shutdown

func (b *Batcher) Shutdown()

Shutdown waits for all active calls to Add to finish, then returns. After Shutdown is called, all subsequent calls to Add fail. Shutdown should be called only once.

type Options

type Options struct {
	// Maximum number of concurrent handlers. Defaults to 1.
	MaxHandlers int
	// Minimum size of a batch. Defaults to 1.
	MinBatchSize int
	// Maximum size of a batch. 0 means no limit.
	MaxBatchSize int
	// Maximum bytesize of a batch. 0 means no limit.
	MaxBatchByteSize int
}

Options sets options for Batcher.

func (*Options) NewMergedOptions

func (o *Options) NewMergedOptions(opts *Options) *Options

newMergedOptions returns o merged with opts.