README
¶
go-batch
Batch processing and chunking utilities for Go. Generic, concurrent, zero dependencies.
Installation
go get github.com/philiprehberger/go-batch
Usage
Chunking
import "github.com/philiprehberger/go-batch"
records := []Record{...} // 1000 records
for _, chunk := range batch.Chunk(records, 100) {
db.BulkInsert(chunk) // insert 100 at a time
}
Concurrent Processing
import "github.com/philiprehberger/go-batch"
err := batch.Process(ctx, userIDs, 50, func(ctx context.Context, ids []int) error {
return sendNotifications(ctx, ids)
}, batch.WithWorkers(4))
Auto-flushing Accumulator
import "github.com/philiprehberger/go-batch"
acc := batch.NewAccumulator(func(events []Event) {
publishEvents(events)
}, batch.FlushSize[Event](100), batch.FlushInterval[Event](5*time.Second))
for event := range incomingEvents {
acc.Add(event)
}
acc.Stop()
API
| Function | Description |
|---|---|
Chunk[T](items, size) |
Split slice into batches |
Process[T](ctx, items, size, fn, opts...) |
Concurrent batch processing |
WithWorkers(n) |
Set number of concurrent workers |
NewAccumulator[T](fn, opts...) |
Auto-flushing item accumulator |
FlushSize[T](n) |
Flush when N items accumulated |
FlushInterval[T](d) |
Flush on time interval |
Development
go test ./...
go vet ./...
License
MIT
Documentation
¶
Overview ¶
Package batch provides batch processing and chunking utilities for Go.
It includes generic functions for splitting slices into chunks, processing batches concurrently with bounded parallelism, and an auto-flushing accumulator with size and time-based triggers.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Chunk ¶
Chunk splits items into sub-slices of the given size. The last chunk may contain fewer than size elements. Chunk panics if size is less than or equal to zero.
func Process ¶
func Process[T any](ctx context.Context, items []T, batchSize int, fn func(ctx context.Context, batch []T) error, opts ...ProcessOption) error
Process splits items into batches of batchSize and calls fn for each batch. Batches are processed concurrently up to the number of workers configured via WithWorkers (default 1). If any invocation of fn returns an error, all errors are collected and returned as a single combined error. Processing respects context cancellation — remaining batches are skipped when ctx is cancelled.
Types ¶
type Accumulator ¶
type Accumulator[T any] struct { // contains filtered or unexported fields }
Accumulator collects items and flushes them in batches. Flushing can be triggered manually, by reaching a size threshold, or on a time interval. All methods are safe for concurrent use.
func NewAccumulator ¶
func NewAccumulator[T any](fn func(items []T), opts ...AccumulatorOption[T]) *Accumulator[T]
NewAccumulator creates a new Accumulator that calls fn with the buffered items on each flush. Configure automatic flushing with FlushSize and FlushInterval options.
func (*Accumulator[T]) Add ¶
func (a *Accumulator[T]) Add(item T)
Add appends an item to the accumulator. If a FlushSize has been configured and the number of buffered items reaches that threshold, a flush is triggered automatically.
func (*Accumulator[T]) Flush ¶
func (a *Accumulator[T]) Flush()
Flush sends all buffered items to the flush callback and resets the buffer. It is safe to call concurrently.
func (*Accumulator[T]) Stop ¶
func (a *Accumulator[T]) Stop()
Stop stops the interval timer (if running) and flushes any remaining buffered items. After Stop returns the accumulator should not be used.
type AccumulatorOption ¶
type AccumulatorOption[T any] func(*accumulatorConfig)
AccumulatorOption configures the behaviour of an Accumulator.
func FlushInterval ¶
func FlushInterval[T any](d time.Duration) AccumulatorOption[T]
FlushInterval configures the Accumulator to automatically flush on a recurring time interval. A background goroutine is started and runs until Stop is called.
func FlushSize ¶
func FlushSize[T any](n int) AccumulatorOption[T]
FlushSize configures the Accumulator to automatically flush when n items have been accumulated.
type ProcessOption ¶
type ProcessOption func(*processConfig)
ProcessOption configures the behaviour of Process.
func WithWorkers ¶
func WithWorkers(n int) ProcessOption
WithWorkers sets the number of concurrent workers used by Process. The default is 1 (sequential processing). If n is less than 1 it is treated as 1.
Source Files