policy

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, error)

Flush clears all messages stored by this batch policy and applies any batching processors to them. The returned values distinguish three outcomes:

• (batch, nil): a batch is ready to be sent onwards.
• (nil, nil): the policy is currently empty, or the batching processors intentionally filtered every message away. Callers should treat the buffered transactions as successfully consumed (ack).
• (nil, err): the batching processors failed and the batch has been dropped. Callers should treat the buffered transactions as failed (nack) so the source can retry them, rather than passing unprocessed data on.

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.