batcher

Documentation

Overview

Package batcher facilitates task aggregation and execution.

Basic Usage: Instantiate a Batcher, set up its constraints, and then start adding tasks. As tasks accumulate, they are batched together for execution, either when a maximum task count is reached or a specified duration elapses. Results of the executed tasks are communicated asynchronously via channels.

Example: Create a Batcher with a maximum of 10 tasks or a 5-second wait:

`b := batcher.New(10, 5*time.Second, execFunc)`

Add a task and receive its result:

resultChan := make(chan batcher.BatchResult)
b.AddTask(myTask, resultChan)
result := <-resultChan

Key Components:

• `Batcher`: The main component that manages task queueing, aggregation, and execution.
• `BatchResult`: A structure encapsulating the response for a task.
• `taskEntry`: Internal representation of a task and its associated result channel.

Task Duplication: Batcher identifies tasks by content. For multiple identical tasks, each has a unique result channel. This distinction ensures that identical tasks return their results to the appropriate callers.

Index

• type BatchResult
• type Batcher
  • func New[InputType comparable, ResultType interface{}](entries int, delay time.Duration, ...) *Batcher[InputType, ResultType]
  • func (b *Batcher[InputType, ResultType]) AddTask(t InputType, resultChan chan BatchResult[ResultType])

Types

type BatchResult

type BatchResult[ResultType interface{}] struct {
	Result ResultType
	Err    error
}

BatchResult encapsulates the response of a batched task. A task will either have a result or an error, but not both.

type Batcher

type Batcher[InputType comparable, ResultType interface{}] struct {
	// contains filtered or unexported fields
}

Batcher manages the batching and execution of tasks. It collects tasks up to a specified limit (maxEntries) or waits for a defined duration (maxDelay) before triggering a batch execution. The actual task execution logic is provided by the execFunc, which processes tasks and returns their corresponding results. Tasks are queued via the taskChan and stored in pendingTasks until batch execution.

func New

func New[InputType comparable, ResultType interface{}](entries int, delay time.Duration, fn func(inputs []InputType) (map[InputType]ResultType, error)) *Batcher[InputType, ResultType]

New creates and returns a Batcher configured with the specified maxEntries and maxDelay parameters. Upon instantiation, it immediately launches the internal task manager as a goroutine to oversee batch operations. The provided execFunc is used to execute batch requests.

func (*Batcher[InputType, ResultType]) AddTask

func (b *Batcher[InputType, ResultType]) AddTask(t InputType, resultChan chan BatchResult[ResultType])

AddTask adds a new task to the Batcher's queue.