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.

Source Endpoint
      │
      ▼
[Exchange: input Message + Headers + Properties]
      │
      ▼
Processor Node 1  ──── WireTap ──► Side Channel
      │
      ▼
Processor Node 2 (transform / filter / route)
      │
      ▼
Sink Endpoint

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 incoming Message with payload and headers.

  • output — The response Message (optional).

  • headers — Shared metadata across nodes (Headers).

  • properties — Context properties.

Message

A Message carries the application payload with its own headers. The payload can be any serializable Java type.

Endpoint

Each ChannelAdapter produces Endpoint instances. An Endpoint can expose:

  • A Consumer (event-driven or polling) to receive messages from the source.

  • A Producer to send messages to the destination.

FlowNode

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

Type Responsibility

DefaultProcessorNode

Execution of application logic via Processor

PayloadTransformerNode

Payload transformation via PayloadTransformer

MessageTranslatorNode

Message type translation

ExpressionFilterNode

Expression-based filtering

WireTapProcessorNode

Asynchronous forwarding to a secondary channel without interrupting the main flow

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 (e.g. HTTP) react to events on the server thread.

  • Polling consumers (e.g. Kafka, Time) run on dedicated threads managed by ScheduledPollingConsumerService.

  • Concurrent sources (fromSource(…​?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 in a non-blocking manner.

  • 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/adapters).