batch

type EntryValue ¶

type EntryValue interface {
	CleanUp()
}

type GetLimiterFunc ¶

type GetLimiterFunc func(key string) *rate.Limiter

GetLimiterFunc returns the limiter for a specific batch key. The function must return the same value for any given key.

type GetProcessorMetricsRecorderFunc ¶

type GetProcessorMetricsRecorderFunc func() ProcessorMetricsRecorder

GetProcessorMetricsRecorderFunc returns a metrics recorder a Processor can use to record metrics.

type MetricsRecorder ¶

type MetricsRecorder interface {
	// Records the rate limit delay, if any, incurred by the current batch.
	RecordRateLimitDelay(delay time.Duration)

	// Records completion of processing for the current batch.
	RecordBatchCompletion(size int, err error)
}

MetricsRecorder defines an interface for recording batch metrics.

type ProcessFunc ¶

type ProcessFunc func(ctx context.Context, key string, values []interface{}) ([]interface{}, error)

ProcessFunc is the function that processes a batch of values and return results.

An implementation receives an slice of values to process. It must return an slice of results of the same size; each element corresponding to a result for the element at the same index in the values slice. If an entry contains an error, an error will be returned to the caller that submitted the value. The batch function may return an empty or nil slice iff it returns an error.

type Processor ¶

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

Processor provides facilities to batch values in keyed batches and invoke a function to process a batch of values. It guarantees that only a single invocation of the function can be in-flight at any given time for each keyed batch.

func NewProcessor(fn ProcessFunc, options ...ProcessorOption) *Processor

func (p *Processor) Delete(key string)

func (p *Processor) Do(ctx context.Context, key string, value interface{}) (interface{}, error)

func (p *Processor) DoChan(ctx context.Context, key string, value interface{}) <-chan (Result)

type ProcessorMetricsRecorder ¶

type ProcessorMetricsRecorder interface {
	// Records the processing start for a single batch of values for the specified key.
	RecordBatchStart(key string) MetricsRecorder
}

ProcessorMetricsRecorder defines an interface for recording batch processor metrics.

type ProcessorOption ¶

type ProcessorOption func(p *Processor)

ProcessorOption modifies the Process configuration.

func WithBatchLimits(limit rate.Limit, burst int) ProcessorOption

func WithDelayBeforeStart(delay time.Duration) ProcessorOption

func WithGlobalLimiter(limiter *rate.Limiter) ProcessorOption

func WithGlobalLimits(limit rate.Limit, burst int) ProcessorOption

func WithLogger(l logr.Logger) ProcessorOption

func WithMetricsRecorder(metricsRecorder ProcessorMetricsRecorder) ProcessorOption

func WithVerboseLogLevel(level int) ProcessorOption

type Result ¶

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

Result represents the Result of a batched call for a specific value.