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 Config, mgr interop.Manager) (*Batcher, error)
  • func (p *Batcher) Add(part *message.Part) bool
  • func (p *Batcher) CloseAsync()
  • func (p *Batcher) Count() int
  • func (p *Batcher) Flush() *message.Batch
  • func (p *Batcher) UntilNext() time.Duration
  • func (p *Batcher) WaitForClose(timeout time.Duration) error
• type Config
  • func NewConfig() Config
  • func (p Config) IsNoop() bool

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 Config, mgr interop.Manager) (*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) CloseAsync

func (p *Batcher) CloseAsync()

CloseAsync 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() *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.

func (*Batcher) WaitForClose

func (p *Batcher) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the processor has closed down.

type Config

type Config struct {
	ByteSize   int                `json:"byte_size" yaml:"byte_size"`
	Count      int                `json:"count" yaml:"count"`
	Check      string             `json:"check" yaml:"check"`
	Period     string             `json:"period" yaml:"period"`
	Processors []processor.Config `json:"processors" yaml:"processors"`
}

Config contains configuration parameters for a batch policy.

func NewConfig

func NewConfig() Config

NewConfig creates a default PolicyConfig.

func (Config) IsNoop

func (p Config) IsNoop() bool

IsNoop returns true if this batch policy configuration does nothing.