Barrier

A barrier synchronizes flows. The exchange that reaches it waits until the executions of other flows, for the same business key, have completed, and then goes on unchanged.

Use it when a step of one flow must not run before work done by other flows is finished: an order can be dispatched only after the kitchen has prepared it and the payment has been authorized, and those are two flows of their own.

A barrier does not combine exchanges into one. It only makes one exchange wait for the completion of others. To collect many messages into one, see Split / Aggregate.

Basic usage

Pipelite.defineFlow("dispatch")
    .fromSource("queue://dispatch-in")
    .barrier("await-kitchen-and-payment", barrier -> barrier       (1)
        .correlateBy(c -> c.headerName("order-id"))                (2)
        .await("queue://kitchen")                                  (3)
        .await("queue://payment")
        .timeout(Duration.ofMinutes(30)))                          (4)
    .process("dispatch-order", dispatcher)                         (5)
    .build();

1	`.barrier(name, configurator)` is a step like `.process(…)`: it can be followed by further steps.
2	The key that ties the waiting exchange to the executions it waits for: here, the value of the header `order-id`.
3	One `.await(…)` for each flow to wait for. The flow is identified by the `queue://` address it reads from.
4	How long an exchange may wait, counted from its arrival. A timeout is mandatory.
5	Runs once every awaited flow has completed for the same `order-id`.

The flows kitchen and payment need nothing special: they are ordinary flows that read from queue://kitchen and queue://payment. The framework tells the barrier when one of their executions completes.

Where to place it

A barrier acts where it is declared, not at the start of the flow. The exchange goes through the steps before it as usual, waits at the barrier with the payload, headers and properties those steps left on it, and goes on with the step after the barrier. The steps before the barrier are not run again.

Pipelite.defineFlow("dispatch")
    .fromSource("queue://dispatch-in")
    .process("validate", validator)                  // runs first, once
    .process("set-order-id", orderIdHeaderSetter)    // can prepare the key
    .barrier("await-kitchen", barrier -> barrier
        .correlateBy(c -> c.headerName("order-id"))
        .await("queue://kitchen")
        .timeout(Duration.ofMinutes(30)))
    .process("dispatch-order", dispatcher)           // runs when the wait is over
    .build();

The key of the waiting exchange is read when the exchange reaches the barrier, so a previous step can set the header that the key reads.
The key of the awaited flow is read when an exchange enters that flow, before any of its steps: it must already be on the exchange as it arrives there.
The timeout counts from the arrival at the barrier.
A failure in a step before the barrier happens before any wait: the exchange never reaches the barrier. A retry of a step after the barrier resumes at that step, it does not wait again.
While the exchange waits, the execution of this flow is not complete: a flow that awaits this one keeps waiting until the exchange has gone on and finished, or has been given up at the timeout.

The key

The key says which executions an exchange waits for. The same way of reading the key is applied to the exchange that waits and to the exchange processed by the awaited flow, so both must carry what the key reads.

.correlateBy(c -> c.headerName("order-id"))          // the value of a header
.correlateBy(c -> c.expression("Payload.orderId"))   // an expression, see Expressions

correlateBy(…) sets the key shared by every await. It is optional when every await has its own.

An await can have a key of its own, for a flow that identifies the work differently:

.await("queue://kitchen", c -> c.headerName("order-id"))
.await("queue://payment", c -> c.expression("Payload.order.id"))

A key is a single value. To use several values, compose them in a previous step and put the result in a header.
If the key cannot be read from an exchange of the awaited flow, that flow is not affected: the exchange simply does not count for any waiting exchange, which will end at the timeout.

See Expressions for the syntax. Expressions cannot concatenate strings.

When an awaited flow is complete

An execution of an awaited flow counts as completed when it ends without a definitive failure:

An exchange dropped by a filter, or ended by the flow on purpose, is completed.
A failure that is being retried is not completed: the barrier keeps waiting for the outcome of the retry.
A failure that cannot be recovered (retries exhausted, sent to the dead letter queue, or taken over by the error handling of the flow) is a failure: the waiting exchange does not wait any longer and fails at the barrier, so it goes through the error handling of its own flow.

The awaited flow may complete before the exchange arrives at the barrier. The completion is remembered, so the exchange goes on at once. It is remembered for as long as the longest timeout of the barriers that await that flow.

Where the flow resumes

An exchange that waits does not hold a thread: the worker that brought it to the barrier is free to process other exchanges, so any number of exchanges can wait.

When the wait is over, the exchange goes on with the step after the barrier on the workers of its own flow, not on the workers of the flow that completed last. Its messages, headers and properties are the same as when it arrived.

Timeout

If the awaited flows have not all completed within the timeout, the exchange is given up and sent to the dead letter queue (see Error Handling). The entry says which flows and keys it was still waiting for. Waits that expired are noticed periodically, so the exchange is given up shortly after the timeout, within about thirty seconds.

If a flow awaits the flow of an exchange that was given up, it is told that this execution failed, like any other failure.

Completions that arrive after an exchange was given up release nothing.

Restart

Waiting exchanges and remembered completions are kept in the framework’s home directory, so a restart does not lose them: the exchanges that were waiting are still waiting, and the ones whose wait was over are resumed.

Rules checked at startup

The context does not start when a barrier is declared in a way that cannot work:

an await names a queue:// address that no flow reads;
a flow awaits itself;
flows await each other in a cycle (a → b → c → a), so none of them could ever go on.

The check is on flows, not on keys: two flows that await each other are refused even if they use different keys and would never actually block. Only the waits declared by a barrier count: an exchange that a flow sends to the queue of another flow is a new exchange, not a wait.

Mistakes in the declaration itself, such as no await, no timeout, an await without a key, or the same flow awaited twice, are reported when the flow is defined.

Limits

The awaited flow must be identified by a literal queue:// address: placeholders are not accepted, and other kinds of source cannot be awaited.
A waiting exchange must be serializable (messages, headers and properties), because it is kept on disk while it waits. If it is not, the failure is reported at the barrier.
A barrier works within a single running instance of the application. When you run several instances, every exchange with the same key, the one that waits and the ones it awaits, must be routed to the same instance, and each instance must have its own state folder. If a key moves to another instance while an exchange is waiting, that exchange is not released and ends at the timeout. See Running Several Instances.