This version is still in development and is not considered stable yet.

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.

public interface Processor {
    void process(Exchange exchange, ProcessContribution contribution);
}

Usage in DSL:

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

The Processor receives an Exchange that exposes full access to input, output, and headers, plus a ProcessContribution - most processors ignore it, it exists for the rare case one needs to call stopExecution() and end the flow early from inside the processor itself.

PayloadTransformer

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

public interface PayloadTransformer {
    Object transform(PayloadHolder inputPayload);
}

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-orders", "Headers['type'] == 'ORDER'")

The expression uses the engine from the pipelite-expression module - see Expressions for the full syntax reference. Payload and Headers are bound automatically to the current Exchange’s input payload and headers.

WireTap

The WireTap sends a copy of the current message to a secondary endpoint, then the main flow continues with the original message.

.wireTap("audit-log", "slf4j://audit-logger")
Despite "tap" suggesting a fire-and-forget side channel, wireTap(…​) currently runs the secondary delivery synchronously, on the same thread as the main flow, before continuing - a slow or blocking tap target delays the main flow by exactly as much (tracked as a known limitation, issue #47).

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

Synchronous fork to a secondary endpoint (currently blocks the main flow - issue #47)

.toRoute(configurator)

RouterNode / RecipientListRouterNode

Conditional or dynamic routing (see Routing)