batcher

Documentation

Index

• type Batch
  • func NewBatch(m *utils.Message, group string, deflate bool, done utils.Done, clk clock.Clock, ...) *Batch
  • func (b *Batch) Add(m *utils.Message)
  • func (b *Batch) Send(cb func())
• type Batcher
  • func (batcher *Batcher) OnMessage(m *utils.Message, done utils.Done)
  • func (batcher *Batcher) Spawn(id int) utils.Composer
  • func (batcher *Batcher) Workers() int
• type Config

Types

type Batch

type Batch struct {
	Group        string
	Deflate      bool
	Message      *utils.Message
	Buf          *bytes.Buffer
	Writer       io.Writer
	MessageCount uint64     // Current number of messages in the buffer
	Done         utils.Done // Call to pass the message to the next handler
	Timer        *clock.Timer
	Sent         bool
}

Batch groups multiple messages

func NewBatch

func NewBatch(m *utils.Message, group string, deflate bool, done utils.Done,
	clk clock.Clock, timeoutMillis uint, ready chan *Batch) *Batch

NewBatch creates a new instance of Batch

func (*Batch) Add

func (b *Batch) Add(m *utils.Message)

Add merges a new message in the buffer

func (*Batch) Send

func (b *Batch) Send(cb func())

Send the batch of messages to the next handler in the pipeline

type Batcher

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

Batcher allows to merge multiple messages in a single one

func (*Batcher) OnMessage

func (batcher *Batcher) OnMessage(m *utils.Message, done utils.Done)

OnMessage is called when a new message is receive. Add the new message to a batch

func (*Batcher) Spawn

func (batcher *Batcher) Spawn(id int) utils.Composer

Spawn starts a gorutine that can receive: - New messages that will be added to a existing or new batch of messages - A batch of messages that is ready to send (i.e. batch timeout has expired)

func (*Batcher) Workers

func (batcher *Batcher) Workers() int

Workers returns the number of workers

type Config

type Config struct {
	Workers           int
	Deflate           bool
	TimeoutMillis     uint
	Limit             uint64
	MaxPendingBatches uint
}

Config stores the config for a Batcher