Flow Definition
Entry point
FlowDefinitionBuilder builder = Pipelite.defineFlow("flow-name");The flow name must be unique within the PipeliteContext.
fromSource
Defines the source endpoint of the flow. A source is always a URL, with the scheme of a channel adapter.
.fromSource("time://ticker?period=1000")
.fromSource("http://ingress")
.fromSource("kafka://my-topic")
.fromSource("queue://order-processing")Every source can declare what identifies the requests it receives, for the idempotency key that the steps hand to external systems:
.fromSource("queue://orders", (SourceConcurrencyConfigurer c) -> c
.idempotencyKey(k -> k.headerName("request-id")))queue:// is the queue in front of an internal flow: the receiving end of a queue:// hop. Its name must be unique within the PipeliteContext, since queue://order-processing is read by exactly one flow: start() fails if two flows declare the same one. A bare name with no scheme, such as .fromSource("order-processing"), is not a source and is rejected when the flow is defined.
A queue source can also declare ?concurrency=N to have N consumers of the queue process up to N messages at the same time instead of one at a time:
.fromSource("queue://order-processing?concurrency=4")concurrency gives up the order of arrival. When exchanges that have the same key must still be processed one at a time, declare the key with orderedBy, and the others keep running in parallel: see Order and concurrency.
.fromSource("queue://order-processing", (SourceConcurrencyConfigurer c) -> c
.concurrency(4)
.orderedBy(k -> k.headerName("order-id")))Every concurrent source shares one framework-owned worker pool (default 200 threads, override via ConfigurablePipeliteContext.setMaxSourceWorkerPoolSize(int)) instead of getting dedicated threads of its own — total thread usage stays bounded regardless of how many flows declare concurrency. It is rejected at startup on any source that is not a queue (http://, file://, kafka://, time://, …).
process
Adds a generic processing node.
.process("step-name", (exchange, contribution) -> {
String input = exchange.getInputPayloadAs(String.class);
exchange.setOutputPayload(input.trim());
})The second argument is a Processor (functional interface) - it receives the Exchange and a
ProcessContribution, which most processors ignore (it exists for the rare case one needs to call
stopExecution() and end the flow early from inside the processor).
The name ("step-name") is used for logging and tracing.
transformPayload
Pure payload transformation, without direct access to the Exchange.
.transformPayload("to-dto", payload ->
objectMapper.convertValue(payload.getPayloadAs(Map.class), MyDto.class)
)filter
Stops the flow if the expression evaluates to false.
See Expressions for the syntax.
.filter("non-empty", "${body} != null")wireTap
Sends a copy of the message to a secondary endpoint without blocking the flow.
.wireTap("audit", "slf4j://audit-logger")split
Splits a collection payload into independent messages, processes each one through a dedicated segment of steps, and aggregates the results back into a single message before the flow continues. See Split / Aggregate for complete documentation.
.split("split-items", segment -> segment
.process("step-1", processor)
.end())barrier
Makes the exchange wait until other flows have completed their work for the same key, and then lets it go on unchanged. See Barrier for complete documentation.
.barrier("await-kitchen-and-payment", barrier -> barrier
.correlateBy(c -> c.headerName("order-id"))
.await("queue://kitchen")
.await("queue://payment")
.timeout(Duration.ofMinutes(30)))toSink
Defines the destination endpoint of the flow.
After toSink, only withRetry, withErrorChannel, withExceptionHandler, and build are available.
.toSink("slf4j://output")
.toSink("kafka://output-topic")
.toSink("queue://next-flow")A sink is always a URL: the protocol of a channel adapter, or queue://<name> to hand the Exchange to the flow that reads that queue (see queue://). A bare name such as .toSink("output") is rejected when the flow is defined.
A flow does not need a sink. Leave toSink out and the flow simply ends after its last step; a flow that ends with toRoute or toRecipientList hands the Exchange on through them.
toRoute
Adds a routing node. See Routing for complete documentation.
.toRoute(route -> route.dynamic()
.when("Headers['type'] == 'ORDER'").then("queue://order-flow")
.otherwise("queue://default-flow")
.end()
)withRetry
Enables automatic retry on exception, resuming the flow at the step that failed rather than
replaying it from the source. Available after toSink or as an alternative termination after a
processor - mutually exclusive with withErrorChannel/withExceptionHandler.
An exhaustion action is mandatory: the lambda only type-checks once it ends in
onErrorChannel(…) or onExceptionHandler(…).
.withRetry(retry -> retry
.maxAttempts(3)
.backoff(Backoff.exponential(Duration.ofSeconds(2)))
.onErrorChannel(err -> err.toDLQ()))
.build()maxAttempts and backoff are both optional. See Error Handling
for the full reference, including Backoff.linear/Backoff.exponential and onExceptionHandler.
withErrorChannel
Configures a custom error channel - a single hand-off with no retry attempts, unlike the
onErrorChannel(…) that withRetry uses as its own exhaustion action, which reaches the same
target only after attempts are exhausted.
.withErrorChannel(err -> err.toChannel("queue://dead-letter"))
.build()The target is a URL, like every destination in the DSL: queue://<name> for an internal flow, or a
registered channel adapter’s protocol (kafka://…) for an external system. err.toDLQ() targets
the framework’s built-in dead letter queue instead, with no flow to write.