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

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).

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 Consumer (event-driven or polling), which receives messages from the external system and starts the flow.
A URL given to toSink(…) makes it a sink endpoint: the framework asks it for a Producer, which sends the flow’s messages to the external system.

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
   └── Consumer startup (EventDriven or Polling)
4. Runtime message processing
5. context.stop()
   └── Orderly shutdown of all services

Threading Model

Event-driven consumers 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 consumers (File, Time) are driven externally, on a schedule, by ScheduledPollingConsumerService.
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.
WireTap nodes forward to a separate channel synchronously, on the same thread as the main flow, before it continues (issue #47).
The RetryService 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.

SPI — Extensibility

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

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

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).