Architecture

Overview

Pipelite is based on a message-oriented pipeline architecture. Each flow represents a processing path that starts from a source, passes through one or more processors, and ends at a sink. Communication between nodes occurs through Exchange objects that encapsulate the in-transit message.

Core Components

PipeliteContext

The PipeliteContext is the central container of the framework. It is responsible for:

Registering the FlowDefinition objects produced by the DSL.
Instantiating Flow instances from their definitions.
Managing the framework lifecycle (start / stop).
Providing the ChannelAdapterManager and the FlowRegistry.

PipeliteContext context = Pipelite.createContext();
context.registerFlowConfigurationClass(MyFlowConfiguration.class);
context.start();

For simple inline definitions, direct registration via registerFlowDefinition is also supported.

Flow

A Flow is the execution unit of a pipeline. Each flow is associated with a source Endpoint that produces Exchange objects and a chain of FlowNode instances that process them sequentially.

Exchange

The Exchange is the message transport container along the flow. It contains:

input — The Message holding the incoming payload.
output — The Message holding the payload produced by the current step.
headers — Metadata that travels along the flow, independent of the payload (Headers).

The user code sees the Exchange interface. Besides the headers, the framework keeps its own values on the exchange (the return address, the routing slip, the details of a failure) as properties, a separate bag that is not visible through the header methods, so a header can never collide with, or be mistaken for, one of them.

Message

A Message is the container of the application payload. It carries no headers: those belong to the Exchange. The payload can be any serializable Java type.

Endpoint

A ChannelAdapter creates an Endpoint for each URL used in a flow definition. The role of that Endpoint depends on where the URL appears:

A URL given to fromSource(…) makes it a source endpoint: the framework asks it for a Source (an EventSource that pushes, or a PollingSource polled on a schedule), which receives messages from the external system and hands them to the framework. The framework wraps it in its own consumer, which owns the queue, the workers, the backpressure and the durable inbox.
A URL given to toSink(…) (or a route destination) makes it a sink endpoint: the framework asks it for a Producer, which sends the flow’s messages to the external system. The framework wraps it in a node of the flow that applies the flow’s error handling around it.

The same adapter serves both roles - for instance kafka://orders can be read by one flow and written by another - but each Endpoint instance plays only one of them.

A source and a sink are structurally different, not just directionally opposite - a source has an internal queue decoupling intake from processing, a sink does not:

Source and sink endpoint internal structure

The queue is what lets a source accept the next message while the flow is still busy with the previous one; a sink has nothing analogous; sending is a direct call on the thread that reached it, with no decoupling of its own.

FlowNode

FlowNode instances are the nodes in the processing chain. Available types:

Type Responsibility

Type	Responsibility
`DefaultProcessorNode`	Execution of application logic via `Processor`
`PayloadTransformerNode`	Payload transformation via `PayloadTransformer`
`MessageTranslatorNode`	Message type translation
`ExpressionFilterNode`	Expression-based filtering
`WireTapProcessorNode`	Forwarding a copy to a secondary channel - currently synchronous, on the main flow’s own thread (issue #47)
`RouterNode`	Content-based routing with a routing table
`RecipientListRouterNode`	Delivery to multiple recipients simultaneously

DefaultProcessorNode

Execution of application logic via Processor

PayloadTransformerNode

Payload transformation via PayloadTransformer

MessageTranslatorNode

Message type translation

ExpressionFilterNode

Expression-based filtering

WireTapProcessorNode

Forwarding a copy to a secondary channel - currently synchronous, on the main flow’s own thread (issue #47)

RouterNode

Content-based routing with a routing table

RecipientListRouterNode

Delivery to multiple recipients simultaneously

Lifecycle

1. Flow definition via DSL  (FlowDefinition)
2. Registration in PipeliteContext
3. context.start()
   ├── ChannelAdapter resolution (classpath scan)
   ├── Endpoint creation
   ├── FlowNode construction
   └── Source startup (wrapped in an event-driven or a polling consumer)
4. Runtime message processing
5. context.stop()
   └── Orderly shutdown of all services

Threading Model

Event sources built directly on external I/O callbacks (HTTP) react on whatever thread the underlying server (Undertow) hands the request to - no dedicated thread of their own.
Kafka is also event-driven, but with its own dedicated poll-loop thread that continuously reads from the Kafka client and feeds results into the same dispatch mechanism HTTP uses.
Polling sources (File, Time) are driven externally, on a schedule, by the framework.
Concurrent sources (fromSource("queue://…?concurrency=N"), see Flow Definition) dispatch to one shared, bounded worker pool instead of allocating dedicated threads per source, so overall thread usage stays bounded no matter how many flows opt in.
An exchange that waits at a barrier does not hold a thread: it is kept aside, and goes on with the workers of its own flow when the flows it waits for have completed.
WireTap nodes forward to a separate channel synchronously, on the same thread as the main flow, before it continues (issue #47).
The retry service handles retries on a separate thread with access to the FlowExecutionDumpRepository.
Every framework-owned thread is named pipelite-<role>-<flow-name> (e.g. pipelite-kafka-order-ingest, pipelite-pool-3), so a thread dump shows at a glance what each thread belongs to.

Public API and hidden core

The modules are layered so that each kind of user sees only what it needs:

Module Who uses it

Module	Who uses it
`pipelite-dsl`	Application code: the flow DSL, and the `Exchange`, `Headers`, `Processor` and `ExceptionHandler` interfaces the code receives.
`pipelite-spi`	Adapter authors: thirteen types - `ChannelAdapter`, `ChannelConfigurer`, `ContextEventListener`, `Endpoint`, `EndpointURL`, `EndpointProperties`, `Producer`, `PollingSourceConfigurer`, `PollingProperties`, `Source`, `EventSource`, `PollingSource` and `SourceContext`. It depends only on `pipelite-dsl`.
`pipelite-core`	The runtime, in packages named `internal` where they are not API: the context, the exchange implementation, the consumers and their dispatch, the backpressure gate, the durable inbox, the `queue://` channel, the retry and dead-letter machinery.

pipelite-dsl

Application code: the flow DSL, and the Exchange, Headers, Processor and ExceptionHandler interfaces the code receives.

pipelite-spi

Adapter authors: thirteen types - ChannelAdapter, ChannelConfigurer, ContextEventListener, Endpoint, EndpointURL, EndpointProperties, Producer, PollingSourceConfigurer, PollingProperties, Source, EventSource, PollingSource and SourceContext. It depends only on pipelite-dsl.

pipelite-core

The runtime, in packages named internal where they are not API: the context, the exchange implementation, the consumers and their dispatch, the backpressure gate, the durable inbox, the queue:// channel, the retry and dead-letter machinery.

An adapter never sees the queue between a source and its flow, the inbox, or the Exchange implementation: it hands exchanges over through a SourceContext and receives them in a Producer.

SPI — Extensibility

To create a new channel adapter it is sufficient to implement:

public interface ChannelAdapter {
    Endpoint createEndpoint(String url);
}

and an Endpoint that creates a Source and/or a Producer.

The adapter is automatically discovered via classpath scan (file META-INF/pipelite.factories, one protocol=fully.qualified.ClassName entry per adapter - see Custom Channel Adapters).