java

Taskito Java SDK

A typed Java 17+ client over the Taskito Rust core, via a hand-written JNI shell (crates/taskito-java).

Feature-complete: producer + inspection + admin + logs, worker task execution, middleware, JSON/signed/encrypted/MessagePack serializers, dashboard, webhooks, CLI, distributed locks, periodic/cron, and the full workflow engine (DAG, fan-out/fan-in, gates, conditions, sub-workflows, sagas, analysis + visualization, canvas). Also: worker resources (DI), enqueue predicates, a KEDA scaler endpoint, producer batching, in-process autoscaling, observability middleware (Micrometer Observation + Sentry), and a Spring Boot 3 starter. Baseline: Java 17 (--release 17); on JDK 22+ hot byte ops take a Panama (FFM) fast path automatically. The native library is bundled per platform — no separate install.

Install

// Gradle
implementation("org.byteveda:taskito:0.18.0")
annotationProcessor("org.byteveda:taskito-processor:0.18.0") // compile-time TaskHandler bindings

<!-- Maven -->
<dependency>
  <groupId>org.byteveda</groupId>
  <artifactId>taskito</artifactId>
  <version>0.18.0</version>
</dependency>

<!-- Maven: the processor is wired through the compiler plugin, not a dependency -->
<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-compiler-plugin</artifactId>
  <configuration>
    <annotationProcessorPaths>
      <path>
        <groupId>org.byteveda</groupId>
        <artifactId>taskito-processor</artifactId>
        <version>0.18.0</version>
      </path>
    </annotationProcessorPaths>
  </configuration>
</plugin>

Companion artifacts: org.byteveda:taskito-test (in-memory backend for unit tests) and org.byteveda:taskito-spring (Boot 3 starter).

Migration

0.18 — source-breaking (pre-1.0): the client interface was renamed and the name Queue now denotes a single named queue.

• The client you open is now Taskito, not Queue: Taskito client = Taskito.builder()…open(); (was Queue queue = …). Taskito.builder() is unchanged.
• Queue is a per-queue handle from Taskito.queue(name), exposing pause() / resume() / isPaused().
• client.pauseQueue("emails") → client.queue("emails").pause() (likewise resume). listPausedQueues() stays on the client as the global view.

Usage

Enqueue

import com.fasterxml.jackson.core.type.TypeReference;
import org.byteveda.taskito.Taskito;
import org.byteveda.taskito.model.Job;
import org.byteveda.taskito.model.JobStatus;
import org.byteveda.taskito.model.QueueStats;
import org.byteveda.taskito.task.Task;
import java.util.Map;

// TypeReference preserves generics that a Class token can't; fluent options
// replace the EnqueueOptions builder for the common cases.
Task<Map<String, Object>> sendEmail =
        Task.of("send_email", new TypeReference<Map<String, Object>>() {})
                .queue("emails")
                .priority(5);

try (Taskito taskito = Taskito.builder().sqlite("taskito.db").open()) {
    String id = taskito.enqueue(sendEmail, Map.of("to", "a@b.c"));
    Job job = taskito.getJob(id).orElseThrow();   // job.status == JobStatus.PENDING
    QueueStats stats = taskito.stats();
    taskito.cancel(id);

    // Pause/resume are scoped to one named queue:
    taskito.queue("emails").pause();
    taskito.queue("emails").resume();
}

Taskito.builder() also has .postgres(url) / .redis(url) shortcuts.

Run a worker

import org.byteveda.taskito.events.EventName;
import org.byteveda.taskito.task.Task;
import org.byteveda.taskito.worker.Worker;

Task<Map> add = Task.of("add", Map.class);

Taskito taskito = Taskito.builder().backend("sqlite").url("taskito.db").open();
Worker worker = taskito.worker()
        .handle(add, p -> ((Number) p.get("a")).intValue() + ((Number) p.get("b")).intValue())
        .concurrency(4)
        .on(EventName.SUCCESS, e -> System.out.println("done: " + e.jobId))
        .start();

// Close on SIGTERM/Ctrl-C; awaitShutdown() then unblocks. (Don't put the worker
// in try-with-resources AND call awaitShutdown() inside it — the block can't
// exit to trigger close(), so it would deadlock.)
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    worker.close();
    taskito.close();
}));
worker.awaitShutdown();

Typed tasks from @TaskHandler (compile-time, no reflection)

Annotate handler methods; a compile-time processor generates a <Class>Tasks companion with a typed Task constant per method (full generics, name declared once) plus a bind(...). Add the processor with annotationProcessor("org.byteveda:taskito-processor").

class EmailTasks {
    @TaskHandler("send_email")          // explicit name
    String send(EmailPayload p) { ... }

    @TaskHandler                        // name defaults to "report"
    Report report(List<Metric> metrics) { ... }
}

// generated EmailTasksTasks:
String id = taskito.enqueue(EmailTasksTasks.SEND, payload);

taskito.worker()
        .apply(b -> EmailTasksTasks.bind(b, new EmailTasks()))
        .start();

The annotation is source-retention and the processor emits plain code — zero runtime reflection, GraalVM-native-image friendly.

Workflows

Define a DAG of tasks; steps run in topological order once every predecessor finishes. Attach trackWorkflows() to the worker so node and run state advance as jobs complete.

import org.byteveda.taskito.task.Task;
import org.byteveda.taskito.worker.Worker;
import org.byteveda.taskito.workflows.Workflow;
import org.byteveda.taskito.workflows.WorkflowRun;
import org.byteveda.taskito.workflows.WorkflowStatus;

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

Workflow wf = Workflow.named("etl")
        .step("extract", extract, 1)
        .step("transform", transform, 2, "extract")
        .step("load", load, 3, "transform");

WorkflowRun run = taskito.submitWorkflow(wf);

try (Worker worker = taskito.worker()
        .handle(extract, p -> p * 10)
        .handle(transform, p -> p + 1)
        .handle(load, p -> p)
        .trackWorkflows()
        .start()) {
    WorkflowStatus status = run.await(Duration.ofSeconds(30));
    // status.state == WorkflowState.COMPLETED; status.node("load").get().status
}

A failed step (after its retries) fails the run and skips downstream nodes; run.cancel() skips pending nodes.

Payloads can also be supplied at submit instead of baked into each step — declare structural steps with stepAfter(name, task, deps...) and pass a map:

Workflow etl = Workflow.named("etl")
        .stepAfter("extract", extract)
        .stepAfter("transform", transform, "extract")
        .stepAfter("load", load, "transform");

taskito.submitWorkflow(etl, Map.of("extract", 5, "transform", 6, "load", 7));

A step's effective payload is map.get(name) when present, else the one baked into the step.

Fan-out / fan-in map a step over a producer's result list and gather the results:

Workflow wf = Workflow.named("pipeline")
        .step("seed", seed, 4)                          // returns List.of(1,2,3,4)
        .fanOut("square", square, "each", "seed")       // runs square(x) per item
        .fanIn("sum", sum, "all", "square");            // sum receives [1,4,9,16]

trackWorkflows() advances run state from worker outcomes, so every worker that processes workflow jobs must enable it — in a multi-worker deployment, a run stalls on any node finished by a worker that did not opt in.

Approval gates park a step until it is resolved (or its timeout elapses); register the workflow so the worker holds downstream payloads:

Workflow wf = Workflow.named("deploy")
        .step("build", build, 1)
        .gate("approve", GateConfig.timeout(Duration.ofMinutes(30), GateAction.REJECT), "build")
        .step("ship", ship, 2, "approve");
try (Worker w = taskito.worker().handle(build, ...).handle(ship, ...).trackWorkflows(wf).start()) {
    w.approveGate(run.runId(), "approve");   // or w.rejectGate(runId, "approve", reason)
}

Conditions gate a step on its predecessors' outcomes — Step.of(...).onFailure() / .onSuccess() / .always(), or a callable condition(ctx -> ...). Sub-workflows run a child workflow as a step (Workflow.subWorkflow(name, child, after...)). Sagas roll a failed run back: Step.of(...).compensate(undoTask) compensates completed steps in reverse order, ending the run COMPENSATED.

WorkflowAnalysis (topological order, levels, ancestors/descendants), WorkflowVisualization (Mermaid / DOT), and Canvas (chain/group/chord) round out the engine.

Middleware

taskito.use(new Middleware() {
    @Override public void onEnqueue(EnqueueContext ctx) { /* validate / rewrite */ }
    @Override public void before(TaskContext ctx) { /* trace */ }
    @Override public void onDeadLetter(OutcomeEvent e) { /* alert */ }
});

Dashboard

try (DashboardServer dashboard = DashboardServer.start(queue, 8080, token, staticDir)) {
    // GET /api/stats, /api/jobs, /api/workers, ... ; POST /api/jobs/{id}/cancel, ...
    dashboard.port();
}

Serializers

byte[] key = ...; // 16/24/32 bytes for AES
Taskito secure = Taskito.builder()
        .backend("sqlite").url("taskito.db")
        .serializer(new EncryptedSerializer(new JsonSerializer(), key))
        .open();

Resources (worker dependency injection)

The primary way to use a non-serializable dependency (pool, client, logger) in a handler: register it once and resolve it inside the worker. Scopes: WORKER (built once, shared) and TASK (built + disposed per invocation).

taskito.resource("db", ctx -> openPool());                       // WORKER
taskito.resource("tx", ResourceScope.TASK, ctx -> ctx.<Pool>use("db").begin(), Tx::close);
taskito.worker().handle(save, p -> Resources.<Tx>use("tx").save(p)).start();

When a handler takes @Resource parameters, the @TaskHandler processor wires them from the runtime for you — no Resources.use call needed.

Cross-process references (proxies)

Secondary to resources: when a specific resource identity must travel inside a payload to another process, carry a signed ProxyRef and rebuild it on the worker. Bind an optional TTL and purpose — both are folded into the HMAC.

Proxies proxies = new Proxies(hmacKey).register(new FileProxyHandler());
ProxyRef ref = proxies.deconstruct(file, Duration.ofMinutes(5), "report"); // producer
File same = proxies.resolve(ref, "report");                                // worker (expiry + purpose checked)

Enqueue gates

taskito.predicate("send_email", ctx -> payloadValid(ctx));       // boolean: false → PredicateRejectedException
taskito.gate("send_email", Recipes.businessHours(zone));         // allow / skip / defer / reject
Optional<String> id = taskito.tryEnqueue(emailTask, msg);        // empty when a gate skips
// Recipes: businessHours / timeWindow / dayOfWeek / payloadMatches / featureFlag.

Producer batching

try (Batcher<Event> batcher = Batcher.of(taskito, ingest, 500, Duration.ofMillis(200))) {
    events.forEach(batcher::add);   // flushed in one enqueueMany when full or after the delay
}

Autoscaling + scaler endpoint

taskito.worker().autoscale(AutoscaleOptions.of(2, 32)).handle(task, ...).start();  // resize by depth
try (Scaler scaler = Scaler.start(taskito, ScalerOptions.onPort(9090))) { /* GET /api/scaler for KEDA */ }

Observability + MessagePack (optional deps)

taskito.use(new TaskitoObservation(observationRegistry));   // Micrometer (metrics + tracing)
taskito.use(new SentryMiddleware());                        // report failures to Sentry
Taskito.builder().sqlite("t.db").serializer(new MsgpackSerializer()).open();

io.micrometer:micrometer-observation, io.sentry:sentry, and org.msgpack:jackson-dataformat-msgpack are compileOnly — add the one you use.

Spring Boot 3 starter

Add org.byteveda:taskito-spring; it auto-configures a Taskito bean from taskito.url / taskito.pool-size / taskito.namespace. Define your own Taskito bean to override it.

Structure

Packages are organized by feature; the root holds only the front door.

org.byteveda.taskito
├── Taskito            client interface + entry — Taskito.builder()...open()
├── Queue              named-queue handle (pause/resume) — Taskito.queue(name)
├── DefaultTaskito     package-private client impl (not exported)
├── NamedQueue         package-private Queue impl (not exported)
├── TaskitoException   unchecked error base type
├── errors/            typed exceptions (Serialization/Crypto, Workflow, Lock,
│                      Configuration, Webhook, Resource, PredicateRejected)
├── task/              Task, TaskFunction, EnqueueOptions
├── model/             Job, JobStatus, QueueStats, DeadJob, JobError,
│                      TaskMetric, WorkerInfo, TaskLog, JobFilter  (read-only views)
├── worker/            Worker runtime (concurrency, autoscale)
├── resources/         worker DI — ResourceRuntime, Resources.use(name), scopes
├── proxies/           signed cross-process references — Proxies, ProxyRef, handlers
├── interception/      enqueue-time arg interception — Interceptor, Interception
├── predicates/        enqueue gates — Predicate, EnqueueGate, EnqueueDecision, Recipes
├── batch/             Batcher — producer-side batching
├── autoscale/         Autoscaler, AutoscaleOptions
├── scaler/            Scaler — KEDA HTTP endpoint
├── locks/             Lock, LockInfo
├── scheduling/        PeriodicTask
├── workflows/         DAG builder, run, status, tracker; gates, conditions,
│                      sub-workflows, sagas, Canvas, analysis + visualization
├── serialization/     Serializer SPI + JsonSerializer default; Signed/Encrypted/Msgpack
├── annotation/        @TaskHandler (source-retention; see :processor)
├── middleware/        Middleware hooks
├── contrib/           observability — TaskitoObservation (Micrometer), SentryMiddleware
├── events/            worker outcome events
├── dashboard/ webhooks/ cli/
├── spi/               QueueBackend — seam between API and the native layer
└── internal/          JNI bindings (NativeQueue, NativeWorkflows, NativeLoader, ...)

Subprojects: :processor (compile-time @TaskHandler), :test-support (taskito-test in-memory backend), :spring (taskito-spring Boot 3 starter), :graalvm-smoke (native-image CI check).

The :processor subproject is a standalone compile-time annotation processor (TaskHandlerProcessor) — it depends on nothing, reading @TaskHandler structurally and emitting plain task companions.

The spi.QueueBackend seam keeps the public API independent of JNI: it can be backed by the native library (default) or an in-memory fake in tests, and leaves room for a future FFM/Panama backend without touching the API.

Build & checks

./gradlew build            # cargoBuild → stage native → compile → test → jar
./gradlew test             # JUnit 5
./gradlew spotlessApply    # format (palantir-java-format)
./gradlew spotlessCheck    # verify formatting
./gradlew checkstyleMain   # static analysis
./gradlew check            # test + spotlessCheck + checkstyle

The Rust shell is built and checked with the workspace tooling:

cargo build -p taskito-java --release --features postgres,redis
cargo fmt -p taskito-java -- --check
cargo clippy -p taskito-java -- -D warnings

Native library resolution

At runtime the platform binary is extracted from the JAR and loaded. For local development against a freshly built library, point the loader at it directly:

-Dtaskito.native.lib=/abs/path/to/target/release/libtaskito_java.so

On JDK 22+ the hot byte ops use a Project Panama (FFM) fast path; older JDKs (or a jar built without the overlay) transparently fall back to JNI. FFM calls a restricted native method, so a future JDK will deny it by default. The jar's Enable-Native-Access: ALL-UNNAMED manifest attribute only covers java -jar; apps that use the SDK as a classpath dependency should launch with --enable-native-access=ALL-UNNAMED to grant access and silence the warning.