Build Java workflows

Use a shared typed client to submit and manage pipelines. A producer describes durable work; workers execute its business stages. The Java demo contains every runnable recipe referenced here and needs only pipelogiq-sdk.

Create a typed pipeline

With the files from getting started, the core operation is:

import io.pipelogiq.sdk.Pipelogiq;
import java.math.BigDecimal;

// Inside an application method using the example Order and stage classes:
try (var client = Pipelogiq.connect("pipelogiq.yml")) {
  var created = client.pipeline("Process order")
      .idempotencyKey("order-123")
      .label("service", "orders")
      .context("source", "checkout")
      .stage(ValidateOrder.class, new Order("123", new BigDecimal("42.50")))
      .stage(RecordOrder.class)
      .send();
  System.out.println(created.id());
}

The annotation on each stage supplies its wire handler name. The default display name is the same name. Use .stage("Validate payment", Handler.class, input) when you need a distinct display name, including repeated uses of one handler. Dependency references use these submitted stage names.

A draft shares its client's connection and needs no separate closing. In a server application, inject or retain one client for the application lifetime rather than opening one per request.

Share contracts with an independent producer

A producer does not need worker implementation classes. Share the agreed input schema and wire names:

import io.pipelogiq.sdk.core.client.StageRef;

StageRef<Order> validate = StageRef.of("ValidateOrder", Order.class);
StageRef<Void> record = StageRef.noInput("RecordOrder");

try (var client = Pipelogiq.connect("pipelogiq.yml")) {
  var created = client.pipeline("External order")
      .idempotencyKey("external-order-123")
      .stage(validate, new Order("123", new java.math.BigDecimal("15.00")))
      .stage(record)
      .send();
  System.out.println(created.id());
}

This fragment uses Pipelogiq and Order from the previous example. IndependentProducerExample.java is a complete version with its own DTO and no imports of worker handlers:

java -cp 'target/classes:target/dependency/*' example.IndependentProducerExample external-001 15.00

Leave the orders worker running. Expect COMPLETED and Recorded: true. A contracts module can hold StageRef constants and records; another language can implement the same JSON contract.

Understand inputs, context and labels

Use NoInputStageHandler for a stage reading context only. There is no need for a dummy input or Void.class in its business method. Use require for a mandatory value and get for an optional one. Parallel branches should write distinct keys and join before consuming each other's results.

Labels are not an extra copy of context. Do not put secrets in labels or logs. Historical JSON property names may mention keywords for protocol compatibility; the public Java API uses labels and has no withKeyword alias.

Observe and cancel a pipeline

Inside a method with an existing pipeline ID:

import java.time.Duration;
import io.pipelogiq.sdk.core.client.PipelineStatus;

try (var client = Pipelogiq.connect("pipelogiq.yml")) {
  var snapshot = client.pipelines().get(123);
  System.out.println(snapshot.status());

  var terminal = client.pipelines().waitForCompletion(123, Duration.ofSeconds(60));
  if (terminal.status() == PipelineStatus.COMPLETED) {
    Boolean recorded = terminal.context("recorded", Boolean.class);
    System.out.println(recorded);
  }
  for (var stage : terminal.stages()) {
    System.out.println(stage.name() + ": " + stage.status() + " / " + stage.lastErrorCode());
  }
}

Replace 123 with your ID. The SDK polls with a bounded deadline and propagates errors. Terminal does not mean successful: a returned pipeline may be failed or cancelled. A wait timeout ends local observation, leaving the durable workflow unchanged.

sendAsync, getAsync, waitForCompletionAsync and other named async methods return CompletableFuture. Keep the client open while they run. Cancelling such a future stops that local request/wait. For server-side cancellation, call client.pipelines().cancel(id) or cancelAsync(id) explicitly.

Return failures that express intent

Expected conditions should return a structured outcome:

Configure retry and deadline policy on the submitted stage. This is the policy used by the runnable inventory recipe:

import io.pipelogiq.sdk.core.builders.StageOptions;
import io.pipelogiq.sdk.core.client.Backoff;
import java.time.Duration;

var retry = new StageOptions()
    .maxRetries(2)
    .retryInterval(Duration.ofSeconds(1))
    .backoff(Backoff.EXPONENTIAL)
    .maxRetryInterval(Duration.ofSeconds(5))
    .retryOnErrorCodes("UPSTREAM_ERROR")
    .timeout(Duration.ofSeconds(15));

Pass it to .stage(FetchInventory.class, retry) for a no-input stage, or .stage(Handler.class, input, retry) for a typed input. The server schedules retry attempts. The handler returns an outcome and does not sleep or implement a retry loop. Duration-based stage policies require whole seconds.

Run the retry and dynamic-stage recipe

The downloaded WorkflowExample.java defines FetchInventory, AppendAudit and AuditOrder. Stop the basic worker, then start the extended worker in Terminal A:

java -cp 'target/classes:target/dependency/*' example.WorkflowExample worker

Terminal B:

java -cp 'target/classes:target/dependency/*' example.WorkflowExample submit workflow-001

The inventory stage deliberately returns UPSTREAM_ERROR on attempt one. A server-scheduled retry succeeds, the order is recorded, and an audit stage is appended. The final output is:

Pipeline <id>: COMPLETED
Inventory available: true
Audit appended: true
Stages: 5

Use a fresh order ID to exercise a fresh retry. The example only updates workflow context; it does not call a real inventory service.

Run a branch-and-join workflow

StageOptions.dependsOn("stageName", ...) and runInParallelWith("stageName", ...) describe server execution policy. They are not Java thread instructions. Give branches explicit unique stage names and reference those names, not Java class names or handler annotations. A join should depend on every branch whose context it needs.

For example, an order workflow can validate first, check stock and customer limits in parallel, then record only after both checks. Implement each branch as a separate stage, write separate context keys and test the graph against the server. Additional policies are runNextIfFailed, failIfOutputEmpty, notifyOnFailure and runAsUser; select them deliberately rather than bypassing a failed business prerequisite by default.

Keep creation and side effects idempotent

Use a stable .idempotencyKey(...) for one business submission. The demo derives it from the order ID. A repeated request returns the same pipeline; a new business action needs a new key. If a create response is lost, use client.pipelines().getByIdempotencyKey(key) to reconcile before accidentally creating a second operation.

Pipeline creation idempotency does not make handler effects exactly once. For a database write, payment or notification, deduplicate using an action-specific business key or stable pipeline/stage identity in the receiving system. An execution ID changes on retry. The pipeline's key alone may be too broad when several stages perform different actions.

Pause for a human decision

A handler can return StageResult.waitingForApproval("Review the order"). The stage waits instead of progressing automatically. An authenticated approval service can use the typed client:

// Inside an authorized approval request, with the actual waiting stage ID:
client.stages().approve(stageId);
// Or, on the rejection path:
client.stages().reject(stageId, "Customer cancelled");

These are alternative actions, not two calls to execute together. Verify that the reviewer may decide this exact pipeline/stage. Sending a notification alone does not authorize a mutation. The agent harness exercises waiting, approved and rejected paths without a provider.

Append work dynamically

A producer can append typed definitions through client.pipelines().append(pipelineId, StageDefinition.of(Handler.class, input)).

Inside a handler, return dynamic stages with its result so the server applies them at the stage boundary. The current result API accepts the descriptor's JSON representation; you do not write wire fields yourself:

return StageResult.success("Append audit")
    .append(StageDefinition.of(AuditOrder.class).toJson());

This fragment comes from WorkflowExample.AppendAudit; import io.pipelogiq.sdk.core.client.StageDefinition. The receiving worker must register AuditOrder before it starts. Typed pipeline management and ordinary creation do not need this conversion.

Register and operate a schedule

Use client.schedule(name) with .cron(...), .every(...) or .runOnceAt(...), then .apply(). Schedule registration is an explicit producer operation, separate from worker startup. The schedule guide covers complete commands, time zones, policies, manual triggering and cleanup.