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 consumer that wraps the source and the chain of FlowNode instances. It is created and started automatically by PipeliteContext.start().

Flows can also depend on each other: a step can make the exchange wait until other flows have completed their work for the same key. See Barrier.

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 the event-driven consumers (HTTP, Kafka, Queue, etc.)
       └─ Start the polling consumers (Time, File, etc.)

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