First Pipeline

Standalone example

The following example creates a pipeline that triggers every second (time adapter) and logs the current payload via SLF4J.

import io.pipelite.core.Pipelite;
import io.pipelite.core.context.PipeliteContext;
import io.pipelite.dsl.definition.FlowDefinition;

public class HelloPipelite {

    public static void main(String[] args) {

        FlowDefinition flow = Pipelite.defineFlow("hello-flow")
            .fromSource("time://ticker?period=1000")           (1)
            .process("log-step", (exchange) -> {               (2)
                System.out.println("Tick: " + exchange.getInputPayload());
                exchange.setOutputPayload("processed");
            })
            .toSink("slf4j://output-logger")                   (3)
            .build();                                          (4)

        PipeliteContext context = Pipelite.createContext();    (5)
        context.registerFlowDefinition(flow);
        context.start();                                       (6)
    }
}

1	The `time://` adapter generates a tick every 1000 ms.
2	An inline processor that receives the Exchange and processes the payload.
3	The `slf4j://` sink logs the output message via SLF4J.
4	`build()` returns the immutable `FlowDefinition`.
5	The `PipeliteContext` is the runtime container.
6	`start()` starts all consumers and begins processing.

Spring Boot example

With the Spring Boot starter, flows are declared in a @FlowConfiguration class and activated via @EnablePipelite:

@SpringBootApplication
@EnablePipelite(flowConfigurations = IngressFlowConfiguration.class)   (1)
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

1	`flowConfigurations` lists the classes containing `@DefineFlow` methods.

@FlowConfiguration
public class IngressFlowConfiguration {

    @DefineFlow
    public FlowDefinition ingressFlow() {
        return Pipelite.defineFlow("ingress-flow")
            .fromSource("http://ingress")                 (1)
            .wireTap("audit-log", "slf4j://audit")        (2)
            .process("validate", exchange -> {
                // validation logic
                exchange.setOutputPayload(exchange.getInputPayload());
            })
            .toSink("slf4j://main-log")
            .build();
    }
}

1	The `http://ingress` source opens an HTTP listener on port 80 at path `/ingress`.
2	`wireTap` sends a copy of the message to the audit channel without blocking the main flow.

For full details on the FlowConfiguration SPI — including dependency injection and multiple flow classes — see FlowConfiguration SPI.

Adding error handling

FlowDefinition flow = Pipelite.defineFlow("resilient-flow")
    .fromSource("kafka://my-topic")
    .process("transform", exchange -> {
        // processing that may throw exceptions
    })
    .toSink("slf4j://output")
    .withRetryChannel()                                        (1)
    .build();

1	`withRetryChannel()` enables automatic retry on unhandled exceptions.

For a custom error channel use withErrorChannel(cfg → …). For full details see Error Handling.

Multi-channel concurrent example

pipelite-examples/pipelite-fooddelivery-application is a complete Spring Boot application showing several flows linked together, two of them running with concurrency > 1, fed by four different channel adapters:

time:// (order generator)  ──┐
http:// (order intake)       ─┼──► link:// (concurrent kitchen stage)
file:// (partner orders)    ──┘             │
                                             ▼
                                        kafka:// (fan-out)
                                             │
                                             ▼
                                   link:// (concurrent dispatch stage)
                                             │
                                             ▼
                                   file:// (one output file per courier)

A docker-compose.yml in the module starts a local Kafka broker. Run the app with mvn spring-boot:run and watch the console: a built-in load generator produces sample orders on its own, so concurrent processing is visible immediately with no manual input needed.