policy

Documentation

Overview

Package policy provides tooling for creating and executing Benthos message batch policies.

Index

• func FieldSpec() docs.FieldSpec
• type Batcher
  • func New(conf batchconfig.Config, mgr bundle.NewManagement) (*Batcher, error)
  • func (p *Batcher) Add(part *message.Part) bool
  • func (p *Batcher) Close(ctx context.Context) error
  • func (p *Batcher) Count() int
  • func (p *Batcher) Flush(ctx context.Context) message.Batch
  • func (p *Batcher) UntilNext() time.Duration

Functions

func FieldSpec

func FieldSpec() docs.FieldSpec

FieldSpec returns a spec for a common batching field.

Types

type Batcher

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

Batcher implements a batching policy by buffering messages until, based on a set of rules, the buffered messages are ready to be sent onwards as a batch.

func New

func New(conf batchconfig.Config, mgr bundle.NewManagement) (*Batcher, error)

New creates an empty policy with default rules.

func (*Batcher) Add

func (p *Batcher) Add(part *message.Part) bool

Add a new message part to this batch policy. Returns true if this part triggers the conditions of the policy.

func (*Batcher) Close

func (p *Batcher) Close(ctx context.Context) error

Close shuts down the policy resources.

func (*Batcher) Count

func (p *Batcher) Count() int

Count returns the number of currently buffered message parts within this policy.

func (*Batcher) Flush

func (p *Batcher) Flush(ctx context.Context) message.Batch

Flush clears all messages stored by this batch policy. Returns nil if the policy is currently empty.

func (*Batcher) UntilNext

func (p *Batcher) UntilNext() time.Duration

UntilNext returns a duration indicating how long until the current batch should be flushed due to a configured period. A negative duration indicates a period has not been set.