bufferer

README

Bufferer

Bufferer is a middleware which provides capability for asynchronous data transfer. Bufferer collects items coming from the source endpoint into an in-memory storage and later sends all collected items into the target endpoint as a single batch. Bufferer can dramatically optimize data transfer speed and increase throughput.

How it works

See the code for details. A brief overview of how Bufferer works (in pair with IntervalThrottler) is provided on the picture below. The picture refers to the "old trigger" — this is how Bufferer worked in the past and how it must not be implemented.

Documentation

Index

• func Bufferer(logger log.Logger, config BuffererConfig, r metrics.Registry) func(abstract.Sinker) abstract.AsyncSink
• type Bufferable
• type BuffererConfig

Functions

func Bufferer

func Bufferer(logger log.Logger, config BuffererConfig, r metrics.Registry) func(abstract.Sinker) abstract.AsyncSink

Bufferer buffers AsyncPush operations by storing items in a built-in buffer. The buffer is flushed when any of the following conditions is fulfilled:

• The number of buffered items is greater or equal to config.TriggingCount; OR
• The Value-size of buffered items is greater or equal to config.TriggingSize; OR
• config.TriggingInterval has passed since the start of the last flush; OR
• A non-row item is detected among the incoming items; OR
• The middleware is closed.

Any config property can be set to its default (zero) value, in which case it does not apply.

Types

type Bufferable

type Bufferable interface {
	// BuffererConfig returns a configuration for the bufferer middleware
	BuffererConfig() BuffererConfig
}

Bufferable is an interface for destination configurations.

When implemented, it signals that the destination supports Bufferer as a middleware.

type BuffererConfig

type BuffererConfig struct {
	TriggingCount    int           // items count (of all of the items buffered) triggering a flush
	TriggingSize     uint64        // byte size / Values size (of all of the items buffered) triggering a flush
	TriggingInterval time.Duration // interval triggering a flush, measured from the start of the last flush - which might be caused by other triggers
}