Processors and Transformers

Processing nodes (FlowNode) are the atomic units that compose the logic of a pipeline. They are added via the DSL during the ProcessOperations phase.

Processor

The Processor interface represents generic application logic.

@FunctionalInterface
public interface Processor {
    void process(Exchange exchange) throws Exception;
}

Usage in DSL:

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

The Processor receives an Exchange that exposes full access to input, output, and headers.

PayloadTransformer

PayloadTransformer is a specialization that transforms the payload without needing to explicitly manage the Exchange.

@FunctionalInterface
public interface PayloadTransformer<I, O> {
    O transform(PayloadHolder<I> payload);
}

Usage in DSL:

.transformPayload("to-uppercase", payload ->
    payload.getPayloadAs(String.class).toUpperCase()
)
PayloadTransformer is preferable to Processor when the logic only concerns the payload, with no need to access headers.

Filter (Expression Filter)

The filter node stops the flow from proceeding if the expression evaluates to false.

.filter("only-non-empty", "${body} != null && !${body}.isEmpty()")

The expression uses the engine from the pipelite-expression module. The ${body} placeholder refers to the current payload.

WireTap

The WireTap sends a copy of the current message to a secondary endpoint in a non-blocking manner, without interrupting the main flow.

.wireTap("audit-log", "slf4j://audit-logger")

The message continues to flow normally along the main flow while the copy is sent to the channel identified by the URL.

Message Translator

The MessageTranslatorNode converts the message type from one format to another via a Function<Message, Message>. It is configured internally by the framework during flow construction.

Summary of nodes available via DSL

DSL method Created node Description

.process(name, processor)

DefaultProcessorNode

Generic processing via Processor

.transformPayload(name, transformer)

PayloadTransformerNode

Pure payload transformation

.filter(name, expression)

ExpressionFilterNode

Expression-based filtering

.wireTap(name, url)

WireTapProcessorNode

Non-blocking fork to a secondary endpoint

.toRoute(configurator)

RouterNode / RecipientListRouterNode / RoutingSlipRouterNode

Conditional or dynamic routing (see Routing)