wow-core/me.ahoo.wow.infra.batch/BatchCoordinator

BatchCoordinator

class BatchCoordinator<T : Any>(val name: String, val options: BatchOptions, writer: BatchWriter<T>, keySelector: (T) -> Any = { Unit }, metrics: WowMetrics = WowMetrics.NONE) : GracefullyStoppable

Coordinates bounded, non-blocking admission and graceful shutdown for storage-independent reactive batches.

Each internal batch lane owns buffering and serial writes. The internal admission controller owns the global capacity shared by lanes. Equal keys map to the same serial lane; different lanes may invoke the writer concurrently. This does not impose item order within a native batch. The key selector must be stable and non-blocking; single-lane coordinators do not invoke it.

Accepted results are actively emitted only by owned result tasks. A terminal signal stored before the subscription handshake finishes may subsequently replay on the subscriber thread; caller-provided scheduling is also outside this coordinator's control.

Constructors

BatchCoordinator

constructor(name: String, options: BatchOptions, writer: BatchWriter<T>, keySelector: (T) -> Any = { Unit }, metrics: WowMetrics = WowMetrics.NONE)

Properties

name

val name: String

options

val options: BatchOptions

Functions

open override fun close()

Closes this resource by performing a graceful shutdown with a 30-second timeout.

fun close(timeout: Duration)

stop

open override fun stop()

Synchronously stops this resource with a 30-second timeout.

open override fun stop(timeout: Duration)

Synchronously stops this resource within the specified timeout.

stopGracefully

open override fun stopGracefully(): Mono<Void>

Closes admission and flushes remaining windows. Completion waits for accepted writes and owned notification tasks, including their synchronous callbacks, but not late signal replay or caller-scheduled asynchronous work. Cancelling an observer does not cancel the shared shutdown process.

submit

fun submit(item: T): Mono<Void>

fun submit(itemFactory: () -> T): Mono<Void>

Reserves capacity before invoking the factory, once per subscription. Only successful enqueue accepts a request. The factory runs on the submitting thread and must be finite, non-blocking and free of storage I/O. Closing does not wait for a factory that has not yet enqueued its item; when that factory returns, its reservation is released and enqueue fails.