Workflows

Task<Integer> extract = Task.of("extract", Integer.class);
Task<Integer> transform = Task.of("transform", Integer.class);
Task<Integer> load = Task.of("load", Integer.class);

try (FlexiQ queue = FlexiQ.builder().url("pipeline.db").open()) {
    Workflow etl = Workflow.named("etl")
            .step("extract", extract, 1)
            .step("transform", transform, 2, "extract")
            .step(Step.of("load", load, 3).after("transform").maxRetries(5).build());

    WorkflowRun run = queue.submitWorkflow(etl);

    try (Worker worker = queue.worker()
            .handle(extract, p -> p)
            .handle(transform, p -> p)
            .handle(load, p -> p)
            .trackWorkflows()
            .start()) {
        WorkflowStatus status = run.await(Duration.ofSeconds(30)); // blocks until terminal
        System.out.println(status.state);                          // COMPLETED | FAILED | ...
        status.nodes.forEach(n -> System.out.println(n.nodeName + " " + n.status));
    }
}