batcher

Documentation

Index

• type Batcher
  • func New(maxSize int, batchTime time.Duration) *Batcher
  • func (b *Batcher) Await() <-chan time.Time
  • func (b *Batcher) Increment()
  • func (b *Batcher) Reset()

Types

type Batcher

type Batcher struct {
	MaxSize   int
	BatchTime time.Duration
	// contains filtered or unexported fields
}

func New

func New(maxSize int, batchTime time.Duration) *Batcher

New returns a simple Batcher that can be used to buffer & batch values as simple as this:

b := batcher.New(1024, 5*time.Second)
for {
  select {
    case <- b.Await():
      b.Reset()
      flush(messages)
      messages = nil

    case msg <- kafka.Messages():
      b.Increment()
      messages = append(messages, msg)
  }
}

func (*Batcher) Await

func (b *Batcher) Await() <-chan time.Time

Await the Batcher flush.

func (*Batcher) Increment

func (b *Batcher) Increment()

Increment records a new message in the batch.

func (*Batcher) Reset

func (b *Batcher) Reset()

Reset the Batcher after flushing.