functional-reactive
Made in the European Union

Async pipelines with CompletableFutureQueue

Build async pipelines as values, hand them around, and apply them when the input arrives — without writing the same supplyAsync/thenComposeAsync ladder twice.

May 17, 2026

The natural way to chain async work in Java is a ladder of thenComposeAsync calls:

Function<String, CompletableFuture<Pair<String, Integer>>> f = value ->
    CompletableFuture.completedFuture(value)
        .thenComposeAsync(v -> CompletableFuture.supplyAsync(() -> Integer.parseInt(v)))
        .thenComposeAsync(v -> CompletableFuture.supplyAsync(() -> v + " next"))
        .thenComposeAsync(v -> CompletableFuture.supplyAsync(() -> new Pair<>(v, v.length())));

It works, but you’ve written the same supplyAsync(() -> stage.apply(v)) wrapper three times. Add two more stages and the boilerplate dominates. CompletableFutureQueue<T, R> extracts that pattern so each stage is a plain Function.

This tutorial walks through why CompletableFutureQueue looks the way it does, and how to use it for the cases that don’t fit the basic CompletableFuture chain.

The shape of an async stage

Look at one stage of the ladder above:

.thenComposeAsync(v -> supplyAsync(() -> step.apply(v)))

The only variable here is step. Everything else is glue. Extract that glue into a method on a wrapper, and every new stage is a one-liner:

public <N> CFQ<T, N> thenCombineAsync(Function<R, N> next) {
    Function<T, CompletableFuture<N>> chained = this.resultFunction
        .andThen(before -> before.thenComposeAsync(v -> supplyAsync(() -> next.apply(v))));
    return new CFQ<>(chained);
}

Pair that with a starting point — define(Function<T, R>) lifts a plain Function into the queue — and a terminator that hands the assembled function back to you:

public Function<T, CompletableFuture<R>> resultFunction() { return resultFunction; }

That is CompletableFutureQueue in three methods. The only extra surface is the constructor (private) and an Executor-aware overload of thenCombineAsync for when you want a stage to run on a specific pool.

Putting it together

import com.svenruppert.functional.reactive.CompletableFutureQueue;

Function<String, Integer> parse  = Integer::parseInt;
Function<Integer, String> stringify = n -> "value=" + n;
Function<String, Pair<String, Integer>> pair = s -> new Pair<>(s, s.length());

// Build the pipeline as a value
Function<String, CompletableFuture<Pair<String, Integer>>> pipeline =
    CompletableFutureQueue
        .define(parse)
        .thenCombineAsync(stringify)
        .thenCombineAsync(pair)
        .resultFunction();

// Apply it whenever you have an input
pipeline.apply("42")
        .thenAccept(System.out::println);          // ("value=42", 8)

Two things to notice:

  1. The pipeline is a Function, not a running computation. Until you call pipeline.apply(...), no thread does any work.
  2. Each stage is a plain Function. No CompletableFuture in the signatures, no supplyAsync wrappers — just String → Integer, Integer → String, etc.

Why this matters

A CompletableFuture.thenComposeAsync chain forces you to either:

  • define the whole pipeline at the call site (verbose and not reusable), or
  • pass around CompletableFuture<T> instances whose computation has already started.

CompletableFutureQueue keeps description and execution separate. You assemble the pipeline at startup, hand it to a component that doesn’t know what the inputs look like yet, and the component applies it when its input arrives. Different callers can decorate the same base pipeline differently — one for tests, one for production — without rewriting it.

Per-stage executors

Use the (Function, Executor) overload when a stage needs a specific pool:

ExecutorService ioPool  = Executors.newFixedThreadPool(16);
ExecutorService cpuPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

Function<String, CompletableFuture<Result<Report>>> pipeline =
    CompletableFutureQueue
        .define(this::fetchRaw)                       // I/O — runs on ioPool below
        .thenCombineAsync(this::parse,    cpuPool)
        .thenCombineAsync(this::analyse,  cpuPool)
        .thenCombineAsync(this::report,   cpuPool)
        .resultFunction();

define itself uses the common ForkJoinPool. If your first stage is blocking I/O, wrap it in a supplyAsync(..., ioPool) before handing it to define, or start the queue with a no-op define(Function.identity()) and put the I/O in the first thenCombineAsync(..., ioPool).

Array of identical stages

When you need to apply n transformations of the same shape (e.g. running the same normalization step several times), use thenCombineAsyncFromArray:

Function<String, String>[] steps = new Function[] { trim, lower, dedupe };

Function<String, CompletableFuture<String>> pipeline =
    CompletableFutureQueue
        .<String, String>define(Function.identity())
        .thenCombineAsyncFromArray(steps)
        .resultFunction();

Combining with Result

For domain pipelines, every stage should return a Result<T> so failures travel as values. CompletableFutureQueue<T, Result<R>> works the same way — the only thing that changes is the type parameter on the queue:

Function<String, CompletableFuture<Result<Report>>> pipeline =
    CompletableFutureQueue
        .define((CheckedFunction<String, Raw>)  this::fetchRaw)        // String → Result<Raw>
        .thenCombineAsync(r -> r.flatMap((CheckedFunction<Raw, Parsed>)   this::parse))
        .thenCombineAsync(r -> r.flatMap((CheckedFunction<Parsed, Report>) this::analyse))
        .resultFunction();

Each stage receives a Result from the previous one and short-circuits via flatMap if the previous stage failed.

When not to use it

CompletableFutureQueue is for pipelines whose stages are known in advance. If your branching is data-dependent (stage 2 might be A or B based on stage 1’s result), stick with thenCompose directly — or split into two queues and pick at runtime.

Recap

  • A CompletableFuture.thenComposeAsync ladder repeats the same supplyAsync wrapper for every stage.
  • CompletableFutureQueue.define(...).thenCombineAsync(...).resultFunction() extracts that wrapper.
  • The returned Function<T, CompletableFuture<R>> is just a value — apply it later, decorate it, share it.
  • Pass an Executor to a stage when it needs its own pool.
  • Combine with Result<T> for failure-as-value pipelines.

Next

Edit on GitHub →