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. Accepts a URL with the channel adapter scheme.

.fromSource("time://ticker?period=1000")
.fromSource("http://ingress")
.fromSource("kafka://my-topic")

A source with no channel adapter scheme — typically the receiving end of a link:// hop — can also declare ?concurrency=N to process up to N messages at the same time instead of one at a time:

.fromSource("order-processing?concurrency=4")

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 backed by a channel adapter (http://, file://, kafka://, time://, …).

process

Adds a generic processing node.

.process("step-name", exchange -> {
    String input = exchange.getInputPayloadAs(String.class);
    exchange.setOutputPayload(input.trim());
})

The second argument is a Processor (functional interface). 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())

toSink

Defines the destination endpoint of the flow. After toSink, only withRetryChannel, withErrorChannel, and build are available.

.toSink("slf4j://output")
.toSink("kafka://output-topic")

toRoute

Adds a routing node. See Routing for complete documentation.

.toRoute(route -> route.contentBased()
    .when(new PayloadTypeCondition(OrderEvent.class), "link://order-flow")
    .otherwise("link://default-flow")
    .build()
)

withRetryChannel

Enables automatic retry on exception. Available after toSink or as an alternative termination after a processor.

.withRetryChannel()
.build()

withErrorChannel

Configures a custom error channel.

.withErrorChannel(builder ->
    builder.toSink("slf4j://dead-letter").build()
)
.build()

build

Finalizes the definition and returns an immutable FlowDefinition.

FlowDefinition definition = Pipelite.defineFlow("my-flow")
    .fromSource("time://tick")
    .toSink("slf4j://log")
    .build();

build() does not start the flow. The flow is activated only after PipeliteContext.start().