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

Flow and PipeliteContext

PipeliteContext

The PipeliteContext is the main runtime container. It manages flow definition registration, adapter instantiation, and the overall lifecycle.

Standalone creation

PipeliteContext context = Pipelite.createContext();

Registration and startup

The recommended approach is via the FlowConfiguration SPI, using @FlowConfiguration and @DefineFlow annotations:

context.registerDependency("myService", new MyServiceImpl());
context.registerFlowConfigurationClass(MyFlowConfiguration.class);
context.start();

// ... application execution ...

context.stop();

For simple inline definitions, direct registration is also supported:

context.registerFlowDefinition(flowDefinition);
context.start();

// ... application execution ...

context.stop();

FlowDefinition

The FlowDefinition is the immutable object produced by the DSL. It describes the flow topology (source, processors, sink, error channel) without any execution logic.

The framework uses it as a blueprint to build the runtime Flow at start() time.

FlowDefinition definition = Pipelite.defineFlow("my-flow")
    .fromSource("time://ticker?period=5000")
    .toSink("slf4j://log")
    .build();

FlowDefinition instances are registered in the context either directly via registerFlowDefinition or — preferably — through @FlowConfiguration classes. See FlowConfiguration SPI for the recommended registration pattern.

Flow

The Flow is the runtime instance of a FlowDefinition. It encapsulates the source Consumer and the chain of FlowNode instances. It is created and started automatically by PipeliteContext.start().

Detailed lifecycle

PipeliteContext.start()
  │
  ├─ ChannelAdapterManager.initialize()
  │    └─ scan classpath → load ChannelAdapters
  │
  ├─ For each registered FlowDefinition: build its source/sink endpoints and its FlowNode chain
  │    (processors, transformers, routers), then register the resulting flow
  │
  └─ ServiceManager.startAll()
       ├─ Start EventDrivenConsumerService (HTTP, Queue, etc.)
       └─ Start ScheduledPollingConsumerService (Time, Kafka, etc.)

PipeliteContext.stop()
  └─ ServiceManager.stopAll()
       └─ Orderly shutdown of all consumers