Deprecated: moved into github.com/kitsunium/sdk/internal (ADR 0171), which no consumer imports; update with: go get github.com/kitsunium/sdk/pkg/v1/<family>@latest for each family you import && go mod tidy
README
¶
batcher
Generic coalescing buffer (ADR 0014). Batcher[T] accumulates items via Add
and delivers them to a Sink[T] closure in batches — flushed eagerly at a
MaxItems / MaxWeight cap, on an optional FlushEvery ticker, or on Flush
/ Close.
b := batcher.NewBatcher(deliver, batcher.Config[Event]{
MaxItems: 1000,
FlushEvery: time.Second,
WeightOf: func(e Event) int64 { return int64(len(e.Body)) },
MaxWeight: 1 << 20,
})
_ = b.Add(ctx, evt) // eager flush when a cap is reached
_ = b.Flush(ctx) // deliver the pending batch now
_ = b.Close(ctx) // stop the ticker + final flush
The deliver closure carries every sink-specific concern: a pre-delivery
reorder, a per-batch key, or a byte-vs-count weight. A nil WeightOf means
count-only batching (each item weighs 1, MaxWeight is ignored).
Errors are typed: Add / Flush after Close return batcher.Closed; a
deliver-closure failure is wrapped as batcher.DeliverFailed with the original
cause reachable via errors.Is.
stdlib-only, domain-neutral, kernel-layer. See CLAUDE.md for the full
contract.
Documentation
¶
Overview ¶
Package batcher provides a generic coalescing buffer: items are appended via Add and handed to a deliver closure in batches — flushed eagerly when an item-count or weight cap is reached, on an optional background ticker, or on an explicit Flush / Close.
The primitive is stdlib-only and domain-neutral (ADR 0014). It owns its own time.Ticker directly and does not depend on any other kernel lifecycle primitive. The deliver closure carries every domain-specific concern: a pre-delivery reorder, a per-batch key, or a byte-vs-count weight all live in the caller's Sink and WeightOf, never in the batcher.
Concurrency: Add, Flush and Close are safe to call from multiple goroutines; a mutex guards the pending batch. The batch is swapped out under the lock and the deliver closure runs outside it, so a slow delivery never blocks a producer. Add returns batcher.Closed once Close has run.
Sink serialization (V6): the deliver closure is invoked under a dedicated delivery mutex held only across the call, so two flush paths (a cap-triggered Add racing the FlushEvery ticker, or two cap-triggered Adds) never enter the Sink concurrently. A Sink may therefore assume serial invocation — it can append to a shared slice or write to one connection without its own locking. Cross-batch ordering is still NOT guaranteed: serialization bounds concurrency but not the arrival order of the racing batches. The delivery mutex is separate from the pending-batch mutex, so a slow Sink never blocks a producer from appending into the next batch.
Package batcher — range 0.1.5.* (ADR 0014 kernel/batcher block).
Package batcher — the Config value type, in its own file per the one-exported-struct-per-file convention.
Package batcher — declares the sentinels returned by the Batcher operations. Each var's name equals its errs.Define Reason in SCREAMING_SNAKE form.
Index ¶
Constants ¶
const CodeBatcherClosed errs.Code = 0x00_01_05_01 // 0.1.5.1
CodeBatcherClosed identifies an Add or Flush call made after Close.
const CodeBatcherDeliverFailed errs.Code = 0x00_01_05_02 // 0.1.5.2
CodeBatcherDeliverFailed identifies a deliver closure that returned an error.
Variables ¶
var ( // BatcherClosed is returned by Add (and Flush) once Close has run; the // batcher no longer accepts items and its ticker, if any, has been joined. BatcherClosed = errs.Define(CodeBatcherClosed, "BATCHER_CLOSED", "Batcher is closed", "internal/kernel/batcher: Add/Flush called after Close") // BatcherDeliverFailed wraps a deliver closure error so callers can route on // the typed sentinel; the original cause stays reachable via errors.Is. BatcherDeliverFailed = errs.Define(CodeBatcherDeliverFailed, "BATCHER_DELIVER_FAILED", "Batcher delivery failed", "internal/kernel/batcher: the deliver closure returned an error") )
Functions ¶
This section is empty.
Types ¶
type Batcher ¶
type Batcher[T any] struct { // contains filtered or unexported fields }
Batcher coalesces Add'd items into batches delivered through its Sink. It owns its own flush ticker and guards the pending batch with a mutex, so Add / Flush / Close are safe under concurrent callers (ADR 0014).
func NewBatcher ¶
NewBatcher builds a Batcher over deliver with cfg. When cfg.FlushEvery > 0 it spawns a ticker goroutine that flushes on the interval; the goroutine is joined by Close. A nil cfg.OnError is degraded to a no-op so the flush paths stay branch-free.
func (*Batcher[T]) Add ¶
Add appends item to the pending batch and eagerly flushes when the batch reaches MaxItems or MaxWeight. It returns batcher.Closed if Close has run; a cap-triggered deliver failure is returned wrapped as DeliverFailed.
type Config ¶
type Config[T any] struct { // MaxItems forces a flush once the pending batch reaches this many items; // a non-positive value disables the item-count trigger. MaxItems int // MaxWeight forces a flush once the pending batch's summed WeightOf reaches // it; ignored when WeightOf is nil or MaxWeight is non-positive. MaxWeight int64 // WeightOf reports an item's weight (e.g. its byte size). A nil WeightOf // means count-only batching: every item weighs 1 and MaxWeight is ignored. WeightOf func(item T) int64 // FlushEvery, when positive, spawns a ticker goroutine that flushes the // pending batch on the interval; Close joins it. FlushEvery time.Duration // OnError observes deliver failures seen by the background ticker. A nil // hook degrades to a no-op so the flush paths stay branch-free. OnError func(err error) }
Config tunes a Batcher: the item / weight caps that trigger an eager flush, the per-item weight function, the background flush interval, and the background-error observer. The zero Config is usable — it batches by item count only (each item weighs 1, MaxWeight is ignored) with no caps and no ticker, so items accumulate until an explicit Flush or Close.
Source Files