Queue Channel Adapter

The queue adapter is the entrance of an internal flow: a named queue, inside the PipeliteContext, that one flow reads and that any other flow (or the application) can put an Exchange on. It is how flows are chained.

Availability

Built into pipelite-core: there is no separate module and nothing to add to the classpath.

Usage

// Main flow
Pipelite.defineFlow("main-flow")
    .fromSource("http://ingest")
    .process("enrich", (exchange, contribution) -> { /* enrichment */ })
    .toSink("queue://secondary")               (1)
    .build();

// Secondary flow
Pipelite.defineFlow("secondary-flow")
    .fromSource("queue://secondary")           (2)
    .process("finalize", (exchange, contribution) -> { /* ... */ })
    .toSink("slf4j://output")
    .build();

1	`queue://secondary` as a destination puts the Exchange on the queue named `secondary`.
2	`queue://secondary` as a source declares that this flow reads that queue. The same URL on both sides: what one flow writes to, the other reads from.

URL format

queue://<queue-name>

Parameter Description

Parameter	Description
`queue-name`	The name of the queue. Not the flow’s own `defineFlow(…)` name: the flow name is an identity, the queue name is an address.

queue-name

The name of the queue. Not the flow’s own defineFlow(…) name: the flow name is an identity, the queue name is an address.

Every endpoint is a URL

A flow is always addressed by URL, and queue://<queue name> is the URL of an internal one. The same rule holds everywhere a destination is written - toSink(…), the destinations of toRoute(…) and toRecipientList(…), wireTap(…), toChannel(…) - and for PipeliteContext.supplyExchange(…). A source and a sink are URLs too: fromSource(…) and toSink(…) always take a protocol (and a flow that has nothing to deliver simply leaves toSink out). A bare name is neither a destination, nor a source, nor a sink, and it is rejected when the flow is defined - or, for a value that is only known at runtime, when the endpoint is created or the exchange is delivered.

.fromSource("orders")
// IllegalArgumentException: fromSource("orders"): 'orders' is not a URL - a source always has a protocol.
// Write 'queue://orders' for the queue in front of an internal flow, or the protocol of the channel adapter
// that feeds it (http://, kafka://, ...)

Characteristics

One queue, one flow, N consumers. A queue is read by exactly one flow. To process its Exchanges concurrently, declare concurrency on the source (?concurrency=N, or fromSource("queue://orders", (SourceConcurrencyConfigurer c) → c.concurrency(N))): the N consumers of the queue all run that flow. It is the only source that can declare concurrency - see Flow Definition.
The sender enqueues and moves on: the receiving flow runs on the consumer threads of its own queue, not on the sender’s thread. If the queue is full the sender waits (backpressure), so a slow flow slows its senders down instead of letting the queue grow without limit.
The queue holds the Exchange itself: the receiving flow gets the very instance the sender handed over, not a copy. With the durable inbox on (the default for every source except time:// - durableInbox=false opts one out, see Time Channel Adapter for why time:// inverts that default) a serialized copy is also written to disk before the Exchange is enqueued, so it survives a restart.
The queue lives in memory, inside one PipeliteContext: it is not a broker and does not reach another process. To leave the process, use a channel adapter such as kafka://.
Useful for splitting complex pipelines into modular, reusable flows.

Order and `concurrency`

With concurrency greater than one the exchanges of a queue run in parallel, so two exchanges that arrived one after the other can finish in reverse order. That matters when they are, say, two updates of the same order: the effects of the second would be overwritten by the first. concurrency set to one keeps the order, but processes everything one at a time, even exchanges that have nothing to do with each other.

orderedBy is the middle ground: exchanges that have the same key go through the flow one at a time, in the order they arrived, while exchanges with different keys still run in parallel, within concurrency.

Pipelite.defineFlow("customer-updates")
    .fromSource("queue://customer-updates", (SourceConcurrencyConfigurer c) -> c
        .concurrency(4)
        .orderedBy(k -> k.headerName("customer-id")))    // or k.expression("Payload.customerId")
    .process("update-customer", updater)
    .build();

The key says which exchanges are related, it is not a way to sort them: the order is always the order of arrival. It is read the way the key of a barrier is: a header, or an expression.

The second exchange of a key does not start until the first has really been processed, also when something goes wrong:

Retry. An exchange that failed and is waiting for its next attempt still holds its key: the exchanges of that key that arrive meanwhile wait, and start when it has succeeded or has been given up (sent to the dead letter queue, or taken by the error handling). The exchanges of the other keys go on. Without this, a retry would overwrite what a later exchange had already done.
Barrier. An exchange that waits at a barrier holds its key until it has gone through the rest of the flow.
Restart. A retry or a wait that was pending when the application stopped still holds its key when it starts again, so the exchanges that arrive first do not overtake it. With a retry repository or a barrier repository of your own, they must be able to list what they hold for a flow, or the order across a restart is not kept.

Things to know:

The price of an exchange that keeps failing. It holds back its own key for the whole retry period (the attempts times the backoff), while the other keys flow. If that key has a lot of traffic, its exchanges pile up behind it, and when too many are waiting the source slows down its senders, for every key, like it does when the queue is full. Keep the retry of an ordered flow short, or let the error handling take the failure.
An exchange without the key (the header is missing, or the expression has no value) is processed without ordering, and a warning is logged. The guarantee holds for the exchanges that have a key.
Without concurrency every exchange is already processed one at a time: orderedBy is ignored, with a warning.
Queues only. Kafka sources keep their order within a partition.
One instance. The order is kept by the instance that receives the exchanges. When you run several, route the exchanges with the same key to the same instance, see Running Several Instances.

Startup validation

PipeliteContext.start() checks that every queue:// target is a queue that a flow registered in the same context reads: the destinations of toSink(…), of the routes and the recipient list, of wireTap(…), and the target of toChannel(…), alone or as the exhaustion action of a retry. If one is not - a typo, a rename on one side only - start() throws a ContextValidationException that lists every problem, naming the flow and the DSL construct, before anything is started.

A destination that contains an expression (#{…}) is only known when an exchange is routed, so it is not checked at startup; if it names no flow then, the delivery fails with an IllegalArgumentException. Destinations with any other protocol are not checked either.

Because a queue is read by exactly one flow, start() also fails if two flows declare the same queue:// source. The error names the second flow and the one that read the queue first, and points at concurrency as the way to scale it. Only queues count: http://orders, queue://orders and two flows reading the same Kafka topic can coexist, since what belongs to a flow - its durable inbox, the resumption of its retries - is keyed by the flow’s name and not by the resource of its source.