pipelogiqBuild Java workflows
Pass typed context, branch and join stages, classify failures, require approval, and schedule recurring work.
A pipeline is a durable description of work. Your Java process registers handlers and submits stage definitions; Pipelogiq decides which stage may run next and persists its outcome. Give stages useful display names and keep wire handler names stable across deployments.
This guide extends the project from Get started with Java. Its recipes use deterministic local data so you can inspect orchestration without contacting business systems.
Understand inputs, context and labels
An input belongs to a particular stage. Pass records, POJOs, maps or lists to withAction; the builder serializes them. A Java string becomes a JSON string. At the lower-level API, an input string is already serialized JSON, so prefer the builder when passing application objects.
Context carries values between stages. Read a structured value using context.get("order", Order.class), write using put, and remove using remove. Keys are case-insensitive; use consistent spelling in application code. The worker sends a patch of changes with its result. Avoid parallel branches writing the same context key: independent keys make merges predictable.
Labels classify a pipeline and support filtering and server policies. They are not stage input. Use addLabel("service", "orders"), not legacy keyword APIs. Context marked with addSensitiveContextItem or putSensitive reaches authorized workers, but public server output redacts it. Keep credentials out of labels and log messages.
Run a branch-and-join workflow
Create src/main/java/example/WorkflowRecipes.java. This class uses OrdersApp.options and OrdersApp.waitFor from the getting-started project.
package example;
import io.pipelogiq.sdk.core.api.PipelogiqApiClient;
import io.pipelogiq.sdk.core.builders.PipelineBuilder;
import io.pipelogiq.sdk.core.builders.ScheduleBuilder;
import io.pipelogiq.sdk.core.builders.StageOptions;
import io.pipelogiq.sdk.core.execution.StageResult;
import io.pipelogiq.sdk.http.HttpPipelineRunner;
import java.time.Duration;
import java.util.List;
import java.util.UUID;
public final class WorkflowRecipes {
private static void register(HttpPipelineRunner worker) {
worker.registerInstance("PrepareReport", Void.class, (unused, context) -> {
context.put("reportId", "report-" + context.pipelineId());
return StageResult.success("Report prepared");
});
worker.registerInstance("ReadSales", Void.class, (unused, context) -> {
context.throwIfCancelled();
context.put("salesTotal", 4200);
return StageResult.success("Loaded simulated sales total");
});
worker.registerInstance("ReadCosts", Void.class, (unused, context) -> {
context.throwIfCancelled();
context.put("costTotal", 3100);
return StageResult.success("Loaded simulated cost total");
});
worker.registerInstance("SummarizeReport", Void.class, (unused, context) -> {
Integer sales = context.get("salesTotal", Integer.class);
Integer costs = context.get("costTotal", Integer.class);
if (sales == null || costs == null) {
return StageResult.missingRequiredData("Both report branches must finish");
}
context.put("margin", sales - costs);
return StageResult.success("Report margin: " + (sales - costs));
});
worker.registerInstance("ReviewReport", Void.class, (unused, context) ->
StageResult.waitingForApproval("Review this report before continuing"));
worker.registerInstance("FinishReport", Void.class, (unused, context) -> {
context.put("reviewed", true);
return StageResult.success("Review accepted; demo workflow finished");
});
worker.registerInstance("PlanFollowUp", Void.class, (unused, context) ->
StageResult.success("Follow-up planned").append(
PipelineBuilder.stage("Follow-up", "FinishReport", null, null, false)));
worker.registerInstance("RetryDemo", Void.class, (unused, context) -> {
if (context.attempt() == 1) {
return StageResult.upstreamError("Simulated first-attempt outage");
}
return StageResult.success("Simulated upstream recovered");
});
}
public static void main(String[] args) throws Exception {
if (args.length == 0) {
throw new IllegalArgumentException(
"Use worker, parallel, approval, retry, dynamic, approve <stageId>, "
+ "reject <stageId>, append <pipelineId>, schedule, history, or archive");
}
var options = OrdersApp.options("java-recipes");
if (args[0].equals("worker")) {
try (var worker = new HttpPipelineRunner(options)) {
register(worker);
Runtime.getRuntime().addShutdownHook(new Thread(worker::close));
var lifetime = worker.start();
worker.awaitReady(Duration.ofSeconds(30));
System.out.println("Recipe worker ready");
lifetime.join();
}
return;
}
try (var api = new PipelogiqApiClient(options)) {
switch (args[0]) {
case "approve", "reject" -> {
if (args.length != 2) throw new IllegalArgumentException("A stage ID is required");
boolean approved = args[0].equals("approve");
System.out.println(api.resumeStage(Long.parseLong(args[1]), approved,
approved ? null : "Report rejected by the demo reviewer").toPrettyString());
}
case "append" -> {
if (args.length != 2) throw new IllegalArgumentException("A pipeline ID is required");
System.out.println(api.appendStages(Long.parseLong(args[1]), List.of(
PipelineBuilder.stage("Extra review record", "FinishReport", null, null, false)))
.toPrettyString());
}
case "schedule" -> {
try (var schedule = ScheduleBuilder.create("java-hourly-report", api)
.every(Duration.ofHours(1)).overlapPolicy("Skip").catchup("One", null)
.jitter(Duration.ofSeconds(10))
.withAction("Scheduled report", "PrepareReport")) {
System.out.println(schedule.apply().toPrettyString());
}
long id = api.triggerSchedule("java-hourly-report", null).path("pipelineId").asLong();
System.out.println(OrdersApp.waitFor(api, id).toPrettyString());
}
case "history" -> System.out.println(
api.scheduleRuns("java-hourly-report", null, 20).toPrettyString());
case "archive" -> System.out.println(
api.archiveSchedule("java-hourly-report").toPrettyString());
case "parallel", "approval", "retry", "dynamic" -> {
try (var pipeline = PipelineBuilder.create("Java recipe: " + args[0], api)
.withIdempotencyKey("recipe:" + UUID.randomUUID())
.addLabel("example", "java-recipes")) {
switch (args[0]) {
case "parallel" -> pipeline
.withAction("Prepare", "PrepareReport")
.withAction("Sales", "ReadSales", new StageOptions().dependsOn("Prepare"))
.withAction("Costs", "ReadCosts", new StageOptions().dependsOn("Prepare"))
.withAction("Summary", "SummarizeReport",
new StageOptions().dependsOn("Sales", "Costs"));
case "approval" -> pipeline
.withAction("Review", "ReviewReport")
.withAction("Finish", "FinishReport");
case "retry" -> pipeline.withAction("Retry example", "RetryDemo",
new StageOptions().maxRetries(2).retryInterval(2).backoff("exponential"));
case "dynamic" -> pipeline.withAction("Plan", "PlanFollowUp");
}
long id = pipeline.send().path("id").asLong();
System.out.println("Pipeline ID: " + id);
System.out.println(OrdersApp.waitFor(api, id).toPrettyString());
}
}
default -> throw new IllegalArgumentException("Unknown command: " + args[0]);
}
}
}
}Rebuild using the getting-started command and run the worker in one terminal:
java -cp 'target/classes:target/dependency/*' example.WorkflowRecipes workerIn another terminal with the same API variables, run:
java -cp 'target/classes:target/dependency/*' example.WorkflowRecipes parallelSales and Costs depend on the display name Prepare. Summary depends on both branches. With worker capacity available, the branches may run concurrently and contribute different context keys. The final context should contain margin=1100. The server schedules dependencies; the Java builder only describes them. runInParallelWith is also available for server-supported parallel grouping, but explicit dependencies make this recipe's join clear.
Return failures that express intent
Use a classified result when your handler knows what went wrong:
| Outcome | Result helper | Meaning |
|---|---|---|
| Invalid amount or malformed business data | validationError(message) | Terminal validation failure |
| Upstream service unavailable | upstreamError(message) | Retryable upstream failure |
| Rate limit | rateLimitExceeded(message) | Retryable rate-limit failure |
| Business rule rejected the action | businessRejected(message) | Terminal business rejection |
| Missing prerequisite context | missingRequiredData(message) | Terminal missing-data failure |
| Domain-specific condition | retryableError(message, code) or terminalError(message, code) | Explicit custom classification |
failure(message) leaves retryability unspecified for server policy. Do not rely on an arbitrary thrown exception to mean “try again”: an unclassified exception becomes terminal UNHANDLED_EXCEPTION. The worker classifies IO failures and upstream timeouts separately. Automatically classified exception failures discard partial context changes; a handler-returned failure can intentionally carry a context patch.
Run WorkflowRecipes retry to see a simulated first-attempt upstream error followed by success. This exercises actual server retry scheduling; the example's upstream outage is simulated. Stage policies support fixed, linear or exponential backoff, a maximum interval, jitter and retryOnErrorCodes. Retry limits and worker capacity are independent.
Keep creation and side effects idempotent
withIdempotencyKey deduplicates pipeline creation. Reuse a key for the same business submission, and use a new key for a distinct operation. getPipelineByIdempotencyKey(key) retrieves the corresponding pipeline; the create response includes wasExisting. An incompatible reuse can be rejected by the server, so handle conflicts explicitly.
Delivery is at least once. If a handler charges a card or reserves inventory, the receiving system must deduplicate that action too. Use a stable business-operation key, or a namespaced combination such as "reserve:" + context.pipelineId() + ":" + context.stageId(). Persist that key atomically with the external action, or pass it to a downstream API that supports idempotency.
context.idempotencyKey() is shared by all stages in a pipeline. executionId() identifies a delivery and can change on retry. Neither should be used indiscriminately as the key for every external action. A process-local set is not durable deduplication.
Pause for a human decision
Run WorkflowRecipes approval. The first stage returns waitingForApproval, and the client stops polling when it observes that state. Read the waiting stage's id from the returned stages array. The API resumes a stage ID, not a pipeline ID:
java -cp 'target/classes:target/dependency/*' example.WorkflowRecipes approve 456Replace 456 with the waiting stage's ID. To reject it, use reject 456; the recipe supplies an explicit rejection reason. Fetch the pipeline with OrdersApp get <pipelineId> to see the result. This command-line client is a demonstration of an authorized reviewer; implement identity and authorization checks in the application that exposes approval to end users.
A waiting stage is already persisted; it does not hold a Java thread open. A normal rejected approval prevents its dependent continuation. Advanced failure continuations and independent branches follow server dependency policy.
Append work dynamically
Run WorkflowRecipes dynamic. PlanFollowUp returns a result containing an appended stage. The server accepts that append together with the current stage result, which keeps workflow expansion tied to the accepted execution.
For external expansion, the recipe exposes append <pipelineId>. Try it on the approval workflow while it is waiting, then approve the waiting stage. The extra stage is appended through api.appendStages, a separate API write subject to the pipeline's current state. Do not blindly repeat this write after an uncertain response; inspect the existing pipeline before deciding whether to submit again.
Register and operate a schedule
Run WorkflowRecipes schedule while the recipe worker is online. It upserts the named hourly schedule and manually triggers one run immediately. Manual triggering does not advance the normal schedule cursor. Run history to inspect executions, then archive when finished to prevent future automatic runs.
For other triggers, replace .every(Duration.ofHours(1)) with .cron("0 9 * * 1-5", "Europe/Tallinn") or .runOnceAt(Instant.now().plusSeconds(60)) and import java.time.Instant. Cron has five fields, without a seconds field. Intervals use whole seconds and must be at least 10 seconds. A one-shot schedule archives after firing.
Schedule names are application-scoped and case-insensitive. apply() upserts; an unchanged definition preserves its existing cursor. Overlap policies are Skip, Queue, Replace and Allow; catch-up policies are None, One and All. A timezone such as Europe/Tallinn follows local daylight-saving transitions; UTC avoids those transitions. Automatic dispatch requires an online worker.
For lifecycle and retry-delivery behavior, continue to Java workers. For package and configuration details, use the Java reference.