Workflows

Orchestrate multi-step DAGs in Java with the FlexiQ workflow builder.

Orchestrate multi-step DAGs (directed acyclic graphs — steps as nodes, after edges between them, no cycles) where each step is a registered task and its after list declares its dependencies. The Rust core schedules steps in topological order; a worker that calls trackWorkflows() advances the run as each step settles.

java

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));
    }
}

Each .step(name, task, payload, after...) binds a typed Task and its payload; Step.of(...) opens a builder for per-step overrides (queue, maxRetries, timeoutMs, priority, and the specialised kinds below). You can also declare structural steps with stepAfter(name, task, deps...) and supply payloads at submit time via queue.submitWorkflow(workflow, Map.of("extract", 5, ...)).

A DAG, or ordinary control flow?

A workflow is a graph declared up front, one job per node, distributable and inspectable. When the sequence is just the shape of one function — charge, wait, email — reach for durable steps instead: same replay-safety, written with ordinary control flow, inside a single job.

The worker must opt in with trackWorkflows() — workflow node and run state are driven from that worker's job outcomes. Workflows that use gates, callable conditions, or sub-workflows must be registered on the tracking worker with trackWorkflows(workflow) so the tracker holds their deferred payloads and predicates.

Coming from chained JMS listeners or RabbitMQ routing keys?

Wiring a multi-step pipeline by hand usually means one listener per step and custom state (a status column, a Redis key) to track what's done. A Workflow declares the whole DAG — steps, after dependencies, and retries — once, and the tracker walks the graph instead of your code.

Building WorkflowsThe Step.Builder reference, the build+submit flow, and the node status lifecycle.Fan-out / Fan-inSplit a step's result into parallel children and collect results.ConditionsRun a step only on success, on failure, always, or per a predicate.Approval GatesPause a run until a human or external system approves or rejects.Sub-WorkflowsCompose workflow runs hierarchically — delegate a step to a child workflow.Saga CompensationAutomatically roll back completed steps in reverse order when a run fails.Canvas PrimitivesChain, group, and chord shorthands for common workflow shapes.Incremental RunsReuse a step's result across runs — cache hits skip re-execution within a TTL.Analysis & VisualizationIntrospect a workflow definition's DAG — order, levels, ancestors, and diagrams.

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)); } }