Package 

Interface StreamBatcher

  • 
public interface StreamBatcher<T extends Object>

    Batches a bursty stream of items into batches delivered at controlled intervals.

    Implementations buffer items submitted via enqueue and periodically invoke the registered onBatch handler with the items collected so far (in arrival order). A batch is typically emitted when either:

    • a size threshold is reached, or

    • a debounce window elapses.

    Exact batching policy (batch size, initial window, backoff, etc.) is implementation-defined.

    • Batch delivery: Items are passed to onBatch as an immutable snapshot. The same item is never delivered twice.

    • Ordering: Items in a batch are ordered by arrival time; batches preserve overall order.

    • Start/stop: Processors usually start lazily on the first enqueue and stop with stop.

    • Backpressure:enqueue is suspend and may block if the internal buffer is bounded and full.

    • Thread-safety: All functions are safe to call from multiple coroutines.

    • Coalescing high-frequency events (socket messages, UI updates) into fewer handler invocations.

    • Rate-limiting downstream work (parsing, I/O, UI recomposition).

    • Method Summary

      Modifier and Type Method Description
      abstract Result<Unit> start() Starts the processor if it's not already running.
      abstract Unit onBatch(SuspendFunction3<List<T>, Long, Integer, Unit> handler) Registers the batch handler to be invoked whenever a batch is ready.
      abstract Result<Unit> enqueue(T item) Enqueues a single item for debounced processing.
      abstract Boolean offer(T item) Enqueues a single item for debounced processing.
      abstract Result<Unit> stop() Stops the processor and releases resources.

      • Methods inherited from class java.lang.Object

        clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    • Method Detail

      • start

         abstract Result<Unit> start()

        Starts the processor if it's not already running.

        batcher.start()
    .onFailure { cause -> logger.e(cause) { "Unable to start batcher" } }

      • onBatch

         abstract Unit onBatch(SuspendFunction3<List<T>, Long, Integer, Unit> handler)

        Registers the batch handler to be invoked whenever a batch is ready.

        The handler is called on the processor's internal coroutine context.

        Parameters:
        handler - A suspending function receiving: - List<T>: the batch snapshot in arrival order, - Long: the debounce window applied (milliseconds) for this emission (telemetry; value meaning is implementation-defined), - Int: the number of items emitted in this batch (equals batch.size).

      • enqueue

         abstract Result<Unit> enqueue(T item)

        Enqueues a single item for debounced processing.

        This function suspends if the underlying buffer is full (bounded-capacity implementations), providing natural backpressure. It returns a Result:

        • Result.success(Unit) if the item was accepted,

        • Result.failure(cause) if the processor is closed/stopped or cannot accept the item.

        Implementations may start processing lazily on the first call.

        Parameters:
        item - The item to enqueue.

      • offer

         abstract Boolean offer(T item)

        Enqueues a single item for debounced processing.

        This non-suspending variant returns false when the underlying buffer is full (bounded-capacity implementations).

        if (!batcher.offer(event)) {
    metrics.incrementDropped()
}

      • stop

         abstract Result<Unit> stop()

        Stops the processor and releases resources.

        After calling this, no further batches will be emitted and subsequent enqueue or offer calls should fail with Result.failure. Multiple calls to stop are allowed and should be idempotent.

        batcher.stop()
    .onFailure { cause -> logger.w(cause) { "Unable to stop batcher" } }