Class Backlog<T>

java.lang.Object
io.fluxzero.common.Backlog<T>
Type Parameters:
T - The type of item being buffered and processed.
All Implemented Interfaces:
Monitored<List<T>>
public class Backlog<T> extends Object implements Monitored<List<T>>
A thread-safe batching queue that asynchronously flushes its content to a consumer in configurable batch sizes.

This utility is useful for scenarios where multiple values are being added over time and you want to consume them in batches for efficiency—such as sending messages to a remote system, writing to a log, etc.

Flushes are executed on a single background thread, and results (e.g. completion or failure) are tracked via CompletableFutures. Optional monitors may observe each flushed batch.

Key Features

  • Supports both synchronous and asynchronous consumers
  • Flushes automatically after new items are added
  • Tracks flush progress with CompletableFuture per add
  • Customizable error handling via ErrorHandler
  • Monitoring support via Monitored

Typical Use

Backlog<String> backlog = Backlog.forAsyncConsumer(batch -> {
    return sendToServer(batch); // returns CompletableFuture
});
backlog.add("a", "b", "c");

  • Constructor Details

  • Method Details

    • forConsumer

      public static <T> Backlog<T> forConsumer(ThrowingConsumer<List<T>> consumer)
      Creates a new backlog for a synchronous consumer and default batch size and default logging error handler.

    • forConsumer

      public static <T> Backlog<T> forConsumer(ThrowingConsumer<List<T>> consumer, int maxBatchSize)
      Creates a backlog with custom max batch size and default logging error handler.

    • forConsumer

      public static <T> Backlog<T> forConsumer(ThrowingConsumer<List<T>> consumer, int maxBatchSize, ErrorHandler<List<T>> errorHandler)
      Creates a backlog with custom max batch size and error handler.

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer)
      Creates a backlog for an asynchronous consumer with default max batch size and default logging error handler.

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize)
      Creates a backlog for an asynchronous consumer with custom max batch size and default logging error handler.

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, ErrorHandler<List<T>> errorHandler)
      Creates a backlog for an asynchronous consumer with custom max batch size and error handler.

    • forOrderedAsyncConsumer

      @Deprecated(forRemoval=true) public static <T> Backlog<T> forOrderedAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer)
      Deprecated, for removal: This API element is subject to removal in a future version.
      Use forAsyncConsumer(ThrowingFunction, int, int) with one in-flight batch.

    • forOrderedAsyncConsumer

      @Deprecated(forRemoval=true) public static <T> Backlog<T> forOrderedAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize)
      Deprecated, for removal: This API element is subject to removal in a future version.
      Use forAsyncConsumer(ThrowingFunction, int, int) with one in-flight batch.

    • forOrderedAsyncConsumer

      @Deprecated(forRemoval=true) public static <T> Backlog<T> forOrderedAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, ErrorHandler<List<T>> errorHandler)
      Deprecated, for removal: This API element is subject to removal in a future version.
      Use forAsyncConsumer(ThrowingFunction, int, int, ErrorHandler) with one in-flight batch.

    • forOrderedAsyncConsumer

      @Deprecated(forRemoval=true) public static <T> Backlog<T> forOrderedAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, ToLongFunction<? super T> batchWeight, long maxBatchWeight)
      Deprecated, for removal: This API element is subject to removal in a future version.
      Use forAsyncConsumer(ThrowingFunction, int, ToLongFunction, long, int) with one in-flight batch.

    • forOrderedAsyncConsumer

      @Deprecated(forRemoval=true) public static <T> Backlog<T> forOrderedAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, ToLongFunction<? super T> batchWeight, long maxBatchWeight, Duration batchCollectionDelay)
      Deprecated, for removal: This API element is subject to removal in a future version.
      Use forAsyncConsumer(ThrowingFunction, int, ToLongFunction, long, int, Duration) with one in-flight batch.

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, int maxInFlightBatches)
      Creates a backlog for an asynchronous consumer with a bounded number of in-flight batches. A batch remains in flight until the future returned by the consumer completes. A new batch is dispatched as soon as capacity becomes available.

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, int maxInFlightBatches, ErrorHandler<List<T>> errorHandler)
      Creates a bounded asynchronous backlog with a custom error handler.

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, ToLongFunction<? super T> itemWeight, long maxBatchWeight, int maxInFlightBatches)
      Creates an asynchronous backlog bounded by item count, cumulative item weight and in-flight batches.

      The first item is always admitted, even when it exceeds maxBatchWeight, so an oversized item can make progress as a one-item batch.

      Parameters:
      consumer - asynchronous batch consumer
      maxBatchSize - maximum number of items in one batch
      itemWeight - function that returns the non-negative weight of an item
      maxBatchWeight - maximum cumulative weight, except for one individually oversized item
      maxInFlightBatches - maximum number of consumer batches whose returned future has not completed

    • forAsyncConsumer

      public static <T> Backlog<T> forAsyncConsumer(ThrowingFunction<List<T>, CompletableFuture<?>> consumer, int maxBatchSize, ToLongFunction<? super T> itemWeight, long maxBatchWeight, int maxInFlightBatches, Duration batchCollectionDelay)
      Creates an asynchronous backlog bounded by item count, cumulative item weight and in-flight batches, with a bounded collection delay whenever an idle backlog starts flushing.

      The delay only applies to the first batch after the backlog was idle. Batches already queued behind an active consumer are drained immediately. This allows very short micro-batching windows without delaying a sustained backlog once it has filled.

      Parameters:
      consumer - asynchronous batch consumer
      maxBatchSize - maximum number of items in one batch
      itemWeight - function that returns the non-negative weight of an item
      maxBatchWeight - maximum cumulative weight, except for one individually oversized item
      maxInFlightBatches - maximum number of consumer batches whose returned future has not completed
      batchCollectionDelay - maximum time to collect concurrent items after an idle start

    • add

      @SafeVarargs public final CompletableFuture<Void> add(T... values)
      Adds values to the backlog.
      Parameters:
      values - one or more values to enqueue
      Returns:
      a future that completes when the values are processed by the consumer.

    • add

      public CompletableFuture<Void> add(Collection<? extends T> values)
      Adds a collection of values to the backlog.
      Parameters:
      values - collection of values to enqueue
      Returns:
      a future that completes when the values are processed by the consumer.

    • addUntracked

      public void addUntracked(T value)
      Adds one value without allocating a separate flush future.

      Use this only when the asynchronous consumer owns completion and failure propagation for the value itself. Consumer failures still reach the configured backlog error handler.

    • addAllUntracked

      public void addAllUntracked(Collection<? extends T> values)
      Adds multiple values without creating per-value flush futures.
      See Also:

    • registerMonitor

      public Registration registerMonitor(Consumer<List<T>> monitor)
      Adds a monitor to observe flushed batches.
      Specified by:
      registerMonitor in interface Monitored<T>
      Parameters:
      monitor - the observer
      Returns:
      a Registration that can be used to remove the monitor

    • shutDown

      public void shutDown()
      Shuts down the internal executor service cleanly.