Batcher (size + time coalescing)
Source: Custom
Topics: generics, time.Timer, channels, graceful shutdown
Problem
Implement a generic Batcher[T] that coalesces individual items into batches, flushing when
either a size threshold is reached or a time window elapses. This is the "batch
processing / traffic shaping" pattern a gateway uses to amortize expensive downstream calls
(bulk DB writes, batched API requests, log shipping).
Requirements:
New(maxSize int, maxWait time.Duration, flush func([]T)) *Batcher[T] — start a background worker.Add(item T) — append an item to the current batch.- Flush the batch when it reaches
maxSize, or when maxWait elapses since the first item
of the current batch — whichever comes first.
Close() — flush any pending items, stop the worker, and return only once it has exited.- The
flush callback must never be called with an empty slice.
type Batcher[T any] struct { ... }
func New[T any](maxSize int, maxWait time.Duration, flush func([]T)) *Batcher[T]
func (b *Batcher[T]) Add(item T)
func (b *Batcher[T]) Close()
Key concepts
- Single owner goroutine: the batch slice is owned by one worker goroutine that
selects over
the input channel and a timer — no mutex needed because only that goroutine touches the batch.
- Timer started on first item: reset the timer when a batch goes from empty → 1 item, so the
window measures "time since batch started", not wall-clock ticks.
- Go 1.23+ timer semantics: since Go 1.23,
Timer.Stop/Reset no longer require the
drain-the-channel dance — a stopped timer won't deliver a stale tick. This package relies on that.
- Graceful close: closing the input channel lets the worker flush the tail batch before exiting;
sync.WaitGroup makes Close block until that's done.
Run
go test -v -race ./challenges/concurrency/batcher/
Sign in to submit your solution.