diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index b7e55980..7bb3e3ec 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -127,6 +127,8 @@ pub struct JobView<'a> { pub error: Option<&'a str>, pub unique_key: Option<&'a str>, pub namespace: Option<&'a str>, + /// Opaque metadata blob (JSON the SDK sets, e.g. middleware-injected trace ids). + pub metadata: Option<&'a str>, } impl<'a> From<&'a Job> for JobView<'a> { @@ -148,6 +150,7 @@ impl<'a> From<&'a Job> for JobView<'a> { error: j.error.as_deref(), unique_key: j.unique_key.as_deref(), namespace: j.namespace.as_deref(), + metadata: j.metadata.as_deref(), } } } diff --git a/sdks/java/README.md b/sdks/java/README.md index 6d2ab2f6..b8478661 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -130,6 +130,21 @@ try (Worker worker = queue.worker() 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: + +```java +Workflow etl = Workflow.named("etl") + .stepAfter("extract", extract) + .stepAfter("transform", transform, "extract") + .stepAfter("load", load, "transform"); + +queue.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: diff --git a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java index d890ef04..e6eae07a 100644 --- a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java +++ b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java @@ -83,6 +83,8 @@ private void generate(TypeElement owner, List methods) { } out.append("import com.fasterxml.jackson.core.type.TypeReference;\n") .append("import org.byteveda.taskito.task.Task;\n") + .append("import org.byteveda.taskito.worker.Handler;\n") + .append("import org.byteveda.taskito.worker.HandlerRegistry;\n") .append("import org.byteveda.taskito.worker.Worker;\n\n") .append("/** Generated task descriptors + worker binding for {@link ") .append(ownerSimple) @@ -131,7 +133,27 @@ private void generate(TypeElement owner, List methods) { .append(");\n"); } } - out.append(" return builder;\n }\n}\n"); + out.append(" return builder;\n }\n\n"); + + // handlers() — returns a HandlerRegistry for Worker.Builder.register(...). + out.append(" public static HandlerRegistry handlers(") + .append(ownerSimple) + .append(" impl) {\n return HandlerRegistry.of(\n"); + for (int i = 0; i < methods.size(); i++) { + ExecutableElement method = methods.get(i); + String constant = upperSnake(method.getSimpleName().toString()); + String methodName = method.getSimpleName().toString(); + out.append(" Handler.of(").append(constant).append(", "); + if (isVoid(method)) { + out.append("payload -> {\n impl.") + .append(methodName) + .append("(payload);\n return null;\n })"); + } else { + out.append("impl::").append(methodName).append(")"); + } + out.append(i + 1 < methods.size() ? ",\n" : ");\n"); + } + out.append(" }\n}\n"); write(owner, qualified, out.toString()); } diff --git a/sdks/java/settings.gradle.kts b/sdks/java/settings.gradle.kts index b746be1c..0b6369d4 100644 --- a/sdks/java/settings.gradle.kts +++ b/sdks/java/settings.gradle.kts @@ -1,3 +1,4 @@ rootProject.name = "taskito" include(":processor") +include(":test-support") diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java index 4cc982e3..03548bb5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -86,7 +86,13 @@ private String dispatchEnqueue(String taskName, Object payload, EnqueueOptions o for (Middleware m : middleware) { m.onEnqueue(context); } - return backend.enqueue(taskName, serializer.serialize(context.payload()), encode(context.options())); + EnqueueOptions finalOptions = context.options(); + if (!context.metadata().isEmpty()) { + finalOptions = finalOptions.toBuilder() + .metadata(encode(context.metadata())) + .build(); + } + return backend.enqueue(taskName, serializer.serialize(context.payload()), encode(finalOptions)); } @Override @@ -313,18 +319,31 @@ public long registerPeriodic(PeriodicTask task) { @Override public WorkflowRun submitWorkflow(Workflow workflow) { + return submitWorkflow(workflow, java.util.Collections.emptyMap()); + } + + @Override + public WorkflowRun submitWorkflow(Workflow workflow, Map suppliedPayloads) { List steps = workflow.steps(); Set deferred = deferredNodes(steps); List> specs = new ArrayList<>(steps.size()); // Deferred nodes (fan-out/fan-in + their downstream) have no job — and so - // no payload — at submit; the tracker enqueues them at runtime. + // no payload — at submit; the tracker enqueues them at runtime. A static + // node's payload is the one supplied at submit, else the one baked in. List payloadNames = new ArrayList<>(); List payloads = new ArrayList<>(); for (Step step : steps) { specs.add(stepSpec(step)); if (!deferred.contains(step.name)) { + Object payload = suppliedPayloads.getOrDefault(step.name, step.payload); + // A payloadless structural step (stepAfter) must get its payload at + // submit; fail fast rather than enqueue a null-payload job. + if (payload == null && !suppliedPayloads.containsKey(step.name)) { + throw new TaskitoException("workflow step '" + step.name + + "' has no payload; supply one via submitWorkflow(wf, payloads)"); + } payloadNames.add(step.name); - payloads.add(serializer.serialize(step.payload)); + payloads.add(serializer.serialize(payload)); } } String runId = backend.submitWorkflow( diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java index 6cc7a81d..64db53fa 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -146,6 +146,14 @@ public interface Queue extends AutoCloseable { /** Submit a workflow DAG; returns a handle to the run. */ WorkflowRun submitWorkflow(Workflow workflow); + /** + * Submit a workflow, supplying per-step payloads keyed by step name. A step's + * effective payload is {@code payloads.get(name)} when present, else the + * payload baked into the step. Pairs with the structural + * {@code Workflow.stepAfter(name, task, deps...)} form. + */ + WorkflowRun submitWorkflow(Workflow workflow, Map payloads); + /** Current status of a workflow run, or empty if it no longer exists. */ Optional workflowStatus(String runId); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java index 0f877ac4..7b4bcde6 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java @@ -1,16 +1,19 @@ package org.byteveda.taskito.middleware; +import java.util.HashMap; +import java.util.Map; import org.byteveda.taskito.task.EnqueueOptions; /** * A job being enqueued, passed to {@link Middleware#onEnqueue} before - * serialization. Replace the payload or options to rewrite the job; throw to - * abort the enqueue. + * serialization. Replace the payload or options to rewrite the job, add to + * {@link #metadata()} to travel with the job, or throw to abort the enqueue. */ public final class EnqueueContext { public final String taskName; private Object payload; private EnqueueOptions options; + private final Map metadata = new HashMap<>(); public EnqueueContext(String taskName, Object payload, EnqueueOptions options) { this.taskName = taskName; @@ -18,6 +21,15 @@ public EnqueueContext(String taskName, Object payload, EnqueueOptions options) { this.options = options; } + /** + * Mutable metadata that travels with the job (readable at execution via + * {@code TaskContext.job().metadata()}). When non-empty it becomes the job's + * metadata blob, replacing any set on the options. + */ + public Map metadata() { + return metadata; + } + public Object payload() { return payload; } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/JobInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/JobInfo.java new file mode 100644 index 00000000..30208806 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/JobInfo.java @@ -0,0 +1,38 @@ +package org.byteveda.taskito.middleware; + +import java.util.Map; +import java.util.function.Supplier; + +/** + * The executing job, exposed to {@link Middleware} hooks. {@link #metadata()} is + * loaded lazily (a storage read) only when first accessed, so middleware that + * never reads it pays nothing. + */ +public final class JobInfo { + private final String id; + private final String taskName; + private final Supplier> metadataLoader; + private Map metadata; + + public JobInfo(String id, String taskName, Supplier> metadataLoader) { + this.id = id; + this.taskName = taskName; + this.metadataLoader = metadataLoader; + } + + public String id() { + return id; + } + + public String taskName() { + return taskName; + } + + /** The job's metadata map (e.g. trace ids injected at enqueue); loaded on first call. */ + public Map metadata() { + if (metadata == null) { + metadata = metadataLoader.get(); + } + return metadata; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java index 43d07879..57fa4aa5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java @@ -1,12 +1,37 @@ package org.byteveda.taskito.middleware; -/** Identifies a task as it executes on a worker. */ +import java.util.HashMap; +import java.util.Map; + +/** + * Identifies a task as it executes on a worker. {@link #attributes()} is a + * per-execution scratch map shared across {@code before}/{@code after}/ + * {@code onError} for the same job; {@link #job()} exposes its metadata. + */ public final class TaskContext { public final String jobId; public final String taskName; - public TaskContext(String jobId, String taskName) { + private final Map attributes = new HashMap<>(); + private final JobInfo job; + + public TaskContext(String jobId, String taskName, JobInfo job) { this.jobId = jobId; this.taskName = taskName; + this.job = job; + } + + public TaskContext(String jobId, String taskName) { + this(jobId, taskName, new JobInfo(jobId, taskName, java.util.Collections::emptyMap)); + } + + /** Mutable per-execution scratch shared across this job's middleware hooks. */ + public Map attributes() { + return attributes; + } + + /** The executing job, including its (lazily loaded) metadata. */ + public JobInfo job() { + return job; } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/Job.java b/sdks/java/src/main/java/org/byteveda/taskito/model/Job.java index dbdd1b00..dba8166b 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/model/Job.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/Job.java @@ -23,6 +23,7 @@ public final class Job { public final String error; public final String uniqueKey; public final String namespace; + public final String metadata; @JsonCreator public Job( @@ -41,7 +42,8 @@ public Job( @JsonProperty("progress") Integer progress, @JsonProperty("error") String error, @JsonProperty("uniqueKey") String uniqueKey, - @JsonProperty("namespace") String namespace) { + @JsonProperty("namespace") String namespace, + @JsonProperty("metadata") String metadata) { this.id = id; this.queue = queue; this.taskName = taskName; @@ -58,5 +60,6 @@ public Job( this.error = error; this.uniqueKey = uniqueKey; this.namespace = namespace; + this.metadata = metadata; } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Handler.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Handler.java new file mode 100644 index 00000000..87449264 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Handler.java @@ -0,0 +1,27 @@ +package org.byteveda.taskito.worker; + +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.task.TaskFunction; + +/** A task descriptor paired with the function that handles it. */ +public final class Handler { + private final Task task; + private final TaskFunction function; + + private Handler(Task task, TaskFunction function) { + this.task = task; + this.function = function; + } + + public static Handler of(Task task, TaskFunction function) { + return new Handler<>(task, function); + } + + public Task task() { + return task; + } + + public TaskFunction function() { + return function; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/HandlerRegistry.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/HandlerRegistry.java new file mode 100644 index 00000000..2f9de4fc --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/HandlerRegistry.java @@ -0,0 +1,28 @@ +package org.byteveda.taskito.worker; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * An immutable bundle of {@link Handler}s registered together via + * {@code Worker.Builder.register}. Generated {@code Tasks.handlers(impl)} + * returns one of these. + */ +public final class HandlerRegistry { + private final List> handlers; + + private HandlerRegistry(List> handlers) { + this.handlers = handlers; + } + + public static HandlerRegistry of(Handler... handlers) { + // Copy the varargs array so later caller mutations can't leak in. + return new HandlerRegistry(Collections.unmodifiableList(new ArrayList<>(Arrays.asList(handlers)))); + } + + public List> handlers() { + return handlers; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index 92bdb51e..8524d38d 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -120,6 +120,19 @@ public Builder apply(Consumer customizer) { return this; } + /** Register a single {@link Handler} (a task + its function). */ + public Builder register(Handler handler) { + handlers.put( + handler.task().name(), new RegisteredTask(handler.task().payloadType(), cast(handler.function()))); + return this; + } + + /** Register every handler in a {@link HandlerRegistry} (e.g. a generated {@code XxxTasks.handlers}). */ + public Builder register(HandlerRegistry registry) { + registry.handlers().forEach(this::register); + return this; + } + public Builder queues(String... queues) { this.queues = Arrays.asList(queues); return this; @@ -159,7 +172,8 @@ public Worker start() { concurrency > 0 ? Executors.newFixedThreadPool(concurrency) : Executors.newCachedThreadPool(); Emitter emitter = new Emitter(); listeners.forEach((name, bound) -> bound.forEach(listener -> emitter.on(name, listener))); - WorkerDispatchBridge bridge = new WorkerDispatchBridge(handlers, serializer, executor, emitter, middleware); + WorkerDispatchBridge bridge = + new WorkerDispatchBridge(backend, handlers, serializer, executor, emitter, middleware); WorkerControl control = backend.startWorker(bridge, encodeOptions()); bridge.bind(control); return new Worker(control, executor); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index 1f09a109..da5ec6b2 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -1,5 +1,9 @@ package org.byteveda.taskito.worker; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -7,9 +11,11 @@ import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.middleware.JobInfo; import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.middleware.TaskContext; import org.byteveda.taskito.serialization.Serializer; +import org.byteveda.taskito.spi.QueueBackend; import org.byteveda.taskito.spi.WorkerBridge; import org.byteveda.taskito.spi.WorkerControl; @@ -20,6 +26,10 @@ * and event listeners. */ final class WorkerDispatchBridge implements WorkerBridge { + private static final ObjectMapper JSON = new ObjectMapper(); + private static final TypeReference> MAP = new TypeReference>() {}; + + private final QueueBackend backend; private final Map handlers; private final Serializer serializer; private final ExecutorService executor; @@ -29,11 +39,13 @@ final class WorkerDispatchBridge implements WorkerBridge { private final CompletableFuture control = new CompletableFuture<>(); WorkerDispatchBridge( + QueueBackend backend, Map handlers, Serializer serializer, ExecutorService executor, Emitter emitter, List middleware) { + this.backend = backend; this.handlers = handlers; this.serializer = serializer; this.executor = executor; @@ -57,7 +69,8 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { bound.failJob(token, "no handler registered for task '" + taskName + "'"); return; } - TaskContext context = new TaskContext(jobId, taskName); + JobInfo job = new JobInfo(jobId, taskName, () -> loadMetadata(jobId)); + TaskContext context = new TaskContext(jobId, taskName, job); try { for (Middleware m : middleware) { m.before(context); @@ -105,6 +118,31 @@ private static void dispatch(Middleware m, EventName name, OutcomeEvent event) { } } + /** Lazily load a job's metadata blob into a map (empty on absence/parse failure). */ + private Map loadMetadata(String jobId) { + try { + JsonNode view = backend.getJobJson(jobId) + .map(WorkerDispatchBridge::readTree) + .orElse(null); + JsonNode blob = view == null ? null : view.get("metadata"); + if (blob == null || blob.isNull()) { + return Collections.emptyMap(); + } + String json = blob.asText(); + return json.isEmpty() ? Collections.emptyMap() : JSON.readValue(json, MAP); + } catch (Exception e) { + return Collections.emptyMap(); + } + } + + private static JsonNode readTree(String json) { + try { + return JSON.readTree(json); + } catch (Exception e) { + return null; + } + } + private static String describe(Throwable t) { String message = t.getMessage(); return message == null ? t.getClass().getSimpleName() : t.getClass().getSimpleName() + ": " + message; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java index 56d3c5b8..955d34f4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java @@ -35,6 +35,16 @@ public Workflow step(String name, Task task, T payload, String... after) return step(Step.of(name, task, payload).after(after).build()); } + /** + * Add a structural step (payload supplied at submit via + * {@code submitWorkflow(wf, payloads)}) that runs after {@code deps}. For a + * job priority, use the {@link Step} builder: {@code step(Step.of(name, + * task).priority(p).after(deps).build())}. + */ + public Workflow stepAfter(String name, Task task, String... deps) { + return step(Step.of(name, task).after(deps).build()); + } + /** Add a fully-configured step. */ public Workflow step(Step step) { steps.add(step); diff --git a/sdks/java/src/test/java/org/byteveda/taskito/AnnotationProcessorTest.java b/sdks/java/src/test/java/org/byteveda/taskito/AnnotationProcessorTest.java index 08a5705a..a8ee2628 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/AnnotationProcessorTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/AnnotationProcessorTest.java @@ -36,4 +36,21 @@ void generatedCompanionEnqueuesAndHandles(@TempDir Path dir) throws Exception { } } } + + @Test + @Timeout(30) + void registerViaGeneratedHandlerRegistry(@TempDir Path dir) throws Exception { + try (Queue queue = + Taskito.builder().sqlite(dir.resolve("hr.db").toString()).open()) { + String id = queue.enqueue(GreeterTasks.GREET, "grace"); + CountDownLatch done = new CountDownLatch(1); + try (Worker worker = queue.worker() + .register(GreeterTasks.handlers(new Greeter())) // generated HandlerRegistry + .on(EventName.SUCCESS, event -> done.countDown()) + .start()) { + assertTrue(done.await(20, TimeUnit.SECONDS), "task should complete"); + assertEquals("hello grace", queue.getResult(id, String.class).orElseThrow()); + } + } + } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java index b667eb27..39b2214e 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java @@ -2,6 +2,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; +import com.fasterxml.jackson.core.type.TypeReference; import java.nio.file.Path; import java.time.Duration; import java.util.ArrayList; @@ -24,7 +25,7 @@ class WorkflowFanOutTest { void fansOutPerItemAndFansInResults(@TempDir Path dir) throws Exception { Task seed = Task.of("fo.seed", Integer.class); Task square = Task.of("fo.square", Integer.class); - Task sum = Task.of("fo.sum", List.class); + Task> sum = Task.of("fo.sum", new TypeReference>() {}); try (Queue queue = Taskito.builder().url(dir.resolve("fo.db").toString()).open()) { // seed(4) -> [1,2,3,4]; square each -> [1,4,9,16]; sum all -> 30 @@ -43,10 +44,10 @@ void fansOutPerItemAndFansInResults(@TempDir Path dir) throws Exception { return out; }) .handle(square, x -> x * x) - .handle(sum, (List xs) -> { + .handle(sum, xs -> { int total = 0; - for (Object x : xs) { - total += ((Number) x).intValue(); + for (Integer x : xs) { + total += x; } return total; }) diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSubmitMapTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSubmitMapTest.java new file mode 100644 index 00000000..e6bc8e5d --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSubmitMapTest.java @@ -0,0 +1,50 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Map; +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.WorkflowState; +import org.byteveda.taskito.workflows.WorkflowStatus; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class WorkflowSubmitMapTest { + + @Test + @Timeout(30) + void structuralStepsWithPayloadsSuppliedAtSubmit(@TempDir Path dir) throws Exception { + Task extract = Task.of("sm.extract", Integer.class); + Task transform = Task.of("sm.transform", Integer.class); + Task load = Task.of("sm.load", Integer.class); + try (Queue queue = + Taskito.builder().sqlite(dir.resolve("sm.db").toString()).open()) { + // Structural steps (deps only); payloads supplied at submit. + Workflow etl = Workflow.named("etl") + .stepAfter("extract", extract) + .stepAfter("transform", transform, "extract") + .stepAfter("load", load, "transform"); + + WorkflowRun run = queue.submitWorkflow(etl, Map.of("extract", 5, "transform", 6, "load", 7)); + 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(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + + // Each step received the payload supplied at submit. + String loadJob = status.node("load").orElseThrow().jobId; + assertEquals(7, queue.getResult(loadJob, Integer.class).orElseThrow()); + } + } + } +} diff --git a/sdks/java/test-support/build.gradle.kts b/sdks/java/test-support/build.gradle.kts new file mode 100644 index 00000000..63f38e07 --- /dev/null +++ b/sdks/java/test-support/build.gradle.kts @@ -0,0 +1,52 @@ +// taskito-test: a pure-Java in-memory QueueBackend for fast unit tests (no JNI, +// no disk). Depends on the runtime for the SPI + models; the runtime does NOT +// depend on it (its own tests live here), so there is no cycle. +plugins { + `java-library` + checkstyle + id("com.diffplug.spotless") version "6.25.0" +} + +group = "org.byteveda" +version = "0.16.4" + +java { + sourceCompatibility = JavaVersion.VERSION_11 + targetCompatibility = JavaVersion.VERSION_11 +} + +tasks.withType().configureEach { + options.release.set(11) +} + +repositories { + mavenCentral() +} + +dependencies { + api(project(":")) + + testImplementation(platform("org.junit:junit-bom:5.10.3")) + testImplementation("org.junit.jupiter:junit-jupiter") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") +} + +spotless { + java { + target("src/**/*.java") + palantirJavaFormat("2.50.0") + removeUnusedImports() + trimTrailingWhitespace() + endWithNewline() + } +} + +checkstyle { + toolVersion = "10.21.4" + configFile = file("../config/checkstyle/checkstyle.xml") + isIgnoreFailures = false +} + +tasks.test { + useJUnitPlatform() +} diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java new file mode 100644 index 00000000..c3f9ef18 --- /dev/null +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -0,0 +1,754 @@ +package org.byteveda.taskito.test; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicLong; +import org.byteveda.taskito.TaskitoException; +import org.byteveda.taskito.spi.QueueBackend; +import org.byteveda.taskito.spi.WorkerBridge; +import org.byteveda.taskito.spi.WorkerControl; + +/** + * A pure-Java, in-memory {@link QueueBackend} for fast unit tests — no JNI, no + * disk. It runs jobs on a small polling worker (dispatch → complete/fail → retry + * or dead-letter), and supports inspection, admin, settings, and locks. + * + *

Not supported: workflows (use a native backend). Periodic registration is + * recorded but never fires. Behaviour approximates the core for testing handlers, + * retries, and dead-lettering — it is not the production scheduler. + */ +public final class InMemoryQueueBackend implements QueueBackend { + private static final ObjectMapper JSON = new ObjectMapper(); + private static final String DEFAULT_QUEUE = "default"; + + private final Map jobs = new ConcurrentHashMap<>(); + private final Map settings = new ConcurrentHashMap<>(); + private final Map locks = new ConcurrentHashMap<>(); + private final List> dead = new CopyOnWriteArrayList<>(); + private final Map>> logs = new ConcurrentHashMap<>(); + private final java.util.Set paused = ConcurrentHashMap.newKeySet(); + private final List workers = new CopyOnWriteArrayList<>(); + private final AtomicLong seq = new AtomicLong(); + + private static long now() { + return System.currentTimeMillis(); + } + + // ── Producer ──────────────────────────────────────────────────── + + @Override + public String enqueue(String taskName, byte[] payload, String optionsJson) { + return enqueueOne(taskName, payload, readNode(optionsJson)); + } + + @Override + public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) { + JsonNode array = readNode(optionsJson); + String[] ids = new String[payloads.length]; + for (int i = 0; i < payloads.length; i++) { + JsonNode opts = array.isArray() && i < array.size() ? array.get(i) : null; + ids[i] = enqueueOne(taskName, payloads[i], opts); + } + return ids; + } + + // synchronized so the unique-key scan + insert is atomic against claimNext + // and concurrent enqueues. + private synchronized String enqueueOne(String taskName, byte[] payload, JsonNode opts) { + String uniqueKey = text(opts, "uniqueKey"); + if (uniqueKey != null) { + for (JobRec existing : jobs.values()) { + if (uniqueKey.equals(existing.uniqueKey) + && ("pending".equals(existing.status) || "running".equals(existing.status))) { + return existing.id; + } + } + } + JobRec job = new JobRec(); + job.id = "im-" + seq.incrementAndGet(); + job.taskName = taskName; + job.payload = payload; + job.queue = optText(opts, "queue", DEFAULT_QUEUE); + job.priority = optInt(opts, "priority", 0); + job.maxRetries = optInt(opts, "maxRetries", 0); + job.timeoutMs = optLong(opts, "timeoutMs", 0); + job.uniqueKey = uniqueKey; + job.metadata = text(opts, "metadata"); + job.namespace = text(opts, "namespace"); + long createdAt = now(); + job.createdAt = createdAt; + job.scheduledAt = createdAt + optLong(opts, "delayMs", 0); + jobs.put(job.id, job); + return job.id; + } + + @Override + public Optional getJobJson(String jobId) { + JobRec job = jobs.get(jobId); + return job == null ? Optional.empty() : Optional.of(toJson(jobView(job))); + } + + @Override + public Optional getResult(String jobId) { + JobRec job = jobs.get(jobId); + return job == null ? Optional.empty() : Optional.ofNullable(job.result); + } + + // synchronized so a pending→cancelled transition can't race claimNext's + // pending→running (which is also on the backend monitor). + @Override + public synchronized boolean cancel(String jobId) { + JobRec job = jobs.get(jobId); + if (job != null && "pending".equals(job.status)) { + job.status = "cancelled"; + job.completedAt = now(); + return true; + } + return false; + } + + @Override + public boolean requestCancel(String jobId) { + JobRec job = jobs.get(jobId); + if (job != null && "running".equals(job.status)) { + job.cancelRequested = true; + return true; + } + return false; + } + + @Override + public boolean isCancelRequested(String jobId) { + JobRec job = jobs.get(jobId); + return job != null && job.cancelRequested; + } + + @Override + public void setProgress(String jobId, int progress) { + JobRec job = jobs.get(jobId); + if (job != null) { + job.progress = Math.max(0, Math.min(100, progress)); + } + } + + // ── Inspection ────────────────────────────────────────────────── + + @Override + public String statsJson() { + return toJson(statsFor(null)); + } + + @Override + public String statsByQueueJson(String queue) { + return toJson(statsFor(queue)); + } + + @Override + public String statsAllQueuesJson() { + Map out = new LinkedHashMap<>(); + jobs.values().stream().map(j -> j.queue).distinct().forEach(q -> out.put(q, statsFor(q))); + return toJson(out); + } + + @Override + public String listJobsJson(String filterJson) { + JsonNode filter = readNode(filterJson); + String status = text(filter, "status"); + String queue = text(filter, "queue"); + String task = text(filter, "task"); + int limit = optInt(filter, "limit", Integer.MAX_VALUE); + int offset = optInt(filter, "offset", 0); + List> views = new ArrayList<>(); + for (JobRec job : jobs.values()) { + if (status != null && !status.equals(job.status)) { + continue; + } + if (queue != null && !queue.equals(job.queue)) { + continue; + } + if (task != null && !task.equals(job.taskName)) { + continue; + } + views.add(jobView(job)); + } + int from = Math.min(offset, views.size()); + int to = limit == Integer.MAX_VALUE ? views.size() : Math.min(from + limit, views.size()); + return toJson(views.subList(from, to)); + } + + @Override + public String jobErrorsJson(String jobId) { + return "[]"; + } + + @Override + public String metricsJson(String taskNameOrNull, long sinceMs) { + return "[]"; + } + + @Override + public String listWorkersJson() { + return "[]"; + } + + // ── Admin ─────────────────────────────────────────────────────── + + @Override + public String listDeadJson(long limit, long offset) { + int from = (int) Math.min(offset, dead.size()); + int to = (int) Math.min(from + limit, dead.size()); + return toJson(new ArrayList<>(dead.subList(from, to))); + } + + @Override + public synchronized String retryDead(String deadId) { + for (Map entry : dead) { + if (deadId.equals(entry.get("id"))) { + JobRec source = jobs.get((String) entry.get("originalJobId")); + dead.remove(entry); + JobRec job = new JobRec(); + job.id = "im-" + seq.incrementAndGet(); + job.taskName = (String) entry.get("taskName"); + job.queue = (String) entry.get("queue"); + job.payload = source == null ? new byte[0] : source.payload; + job.maxRetries = source == null ? 0 : source.maxRetries; + // Preserve the rest of the original job's options so a retry + // behaves like the original, not a stripped-down job. + if (source != null) { + job.priority = source.priority; + job.timeoutMs = source.timeoutMs; + job.uniqueKey = source.uniqueKey; + job.metadata = source.metadata; + job.namespace = source.namespace; + } + job.createdAt = now(); + job.scheduledAt = job.createdAt; + jobs.put(job.id, job); + return job.id; + } + } + throw new TaskitoException("no dead-letter entry: " + deadId); + } + + @Override + public boolean deleteDead(String deadId) { + return dead.removeIf(entry -> deadId.equals(entry.get("id"))); + } + + @Override + public long purgeDead(long olderThanMs) { + int before = dead.size(); + dead.removeIf(entry -> ((Number) entry.get("failedAt")).longValue() < olderThanMs); + return before - dead.size(); + } + + @Override + public long purgeCompleted(long olderThanMs) { + List remove = new ArrayList<>(); + for (JobRec job : jobs.values()) { + if ("complete".equals(job.status) && job.completedAt != null && job.completedAt < olderThanMs) { + remove.add(job.id); + } + } + remove.forEach(jobs::remove); + return remove.size(); + } + + @Override + public void pauseQueue(String queue) { + paused.add(queue); + } + + @Override + public void resumeQueue(String queue) { + paused.remove(queue); + } + + @Override + public String listPausedQueuesJson() { + return toJson(new ArrayList<>(paused)); + } + + @Override + public Optional getSetting(String key) { + return Optional.ofNullable(settings.get(key)); + } + + @Override + public void setSetting(String key, String value) { + settings.put(key, value); + } + + @Override + public boolean deleteSetting(String key) { + return settings.remove(key) != null; + } + + @Override + public String listSettingsJson() { + return toJson(new LinkedHashMap<>(settings)); + } + + // ── Logs ──────────────────────────────────────────────────────── + + @Override + public void writeTaskLog(String jobId, String taskName, String level, String message, String extraOrNull) { + Map entry = new LinkedHashMap<>(); + entry.put("id", "log-" + seq.incrementAndGet()); + entry.put("jobId", jobId); + entry.put("taskName", taskName); + entry.put("level", level); + entry.put("message", message); + entry.put("extra", extraOrNull); + entry.put("loggedAt", now()); + logs.computeIfAbsent(jobId, k -> new CopyOnWriteArrayList<>()).add(entry); + } + + @Override + public String getTaskLogsJson(String jobId) { + return toJson(new ArrayList<>(logs.getOrDefault(jobId, new ArrayList<>()))); + } + + // ── Locks ─────────────────────────────────────────────────────── + + @Override + public synchronized boolean acquireLock(String name, String ownerId, long ttlMs) { + LockRec held = locks.get(name); + if (held != null && held.expiresAt > now() && !held.ownerId.equals(ownerId)) { + return false; + } + locks.put(name, new LockRec(name, ownerId, now(), now() + ttlMs)); + return true; + } + + @Override + public synchronized boolean releaseLock(String name, String ownerId) { + LockRec held = locks.get(name); + if (held != null && held.ownerId.equals(ownerId)) { + locks.remove(name); + return true; + } + return false; + } + + @Override + public synchronized boolean extendLock(String name, String ownerId, long ttlMs) { + LockRec held = locks.get(name); + if (held != null && held.ownerId.equals(ownerId)) { + held.expiresAt = now() + ttlMs; + return true; + } + return false; + } + + @Override + public Optional lockInfoJson(String name) { + LockRec held = locks.get(name); + if (held == null || held.expiresAt <= now()) { + return Optional.empty(); + } + Map view = new LinkedHashMap<>(); + view.put("lockName", held.name); + view.put("ownerId", held.ownerId); + view.put("acquiredAt", held.acquiredAt); + view.put("expiresAt", held.expiresAt); + return Optional.of(toJson(view)); + } + + // ── Periodic ──────────────────────────────────────────────────── + + @Override + public long registerPeriodic( + String name, String taskName, String cron, byte[] args, String queue, String timezone, boolean enabled) { + settings.put("periodic:" + name, taskName + "|" + cron); // recorded, never fires + return now(); + } + + // ── Worker ────────────────────────────────────────────────────── + + @Override + public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) { + InMemoryWorker worker = new InMemoryWorker(bridge, parseQueues(optionsJson)); + workers.add(worker); + worker.start(); + return worker; + } + + /** The worker's queue filter from its options JSON; empty = consume every queue. */ + private java.util.Set parseQueues(String optionsJson) { + java.util.Set queues = new java.util.HashSet<>(); + try { + JsonNode node = JSON.readTree(optionsJson).get("queues"); + if (node != null && node.isArray()) { + node.forEach(q -> queues.add(q.asText())); + } + } catch (Exception ignored) { + // Missing or malformed options — fall back to consuming every queue. + } + return queues; + } + + @Override + public void close() { + workers.forEach(InMemoryWorker::stop); + workers.clear(); + } + + // ── Workflows (unsupported in-memory) ─────────────────────────── + + private static T unsupported() { + throw new UnsupportedOperationException( + "the in-memory backend does not support workflows; use a native backend"); + } + + @Override + public String submitWorkflow( + String name, + int version, + String stepsJson, + String[] payloadNames, + byte[][] payloads, + String queueDefault, + String paramsJson, + String[] deferredNames) { + return unsupported(); + } + + @Override + public String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade) { + return unsupported(); + } + + @Override + public Optional getWorkflowStatusJson(String runId) { + return unsupported(); + } + + @Override + public void cancelWorkflowRun(String runId) { + unsupported(); + } + + @Override + public Optional getWorkflowPlanJson(String runId) { + return unsupported(); + } + + @Override + public Optional workflowNodeForJobJson(String jobId) { + return unsupported(); + } + + @Override + public String[] expandFanOut( + String runId, + String parentNode, + String[] childNames, + byte[][] childPayloads, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority) { + return unsupported(); + } + + @Override + public Optional checkFanOutCompletionJson(String runId, String parentNode) { + return unsupported(); + } + + @Override + public String createDeferredJob( + String runId, + String nodeName, + byte[] payload, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority) { + return unsupported(); + } + + @Override + public void cascadeSkipPending(String runId) { + unsupported(); + } + + @Override + public Optional finalizeRunIfTerminal(String runId) { + return unsupported(); + } + + // ── Internals ─────────────────────────────────────────────────── + + /** Claim a dispatchable pending job (status → running); null if none right now. */ + private synchronized JobRec claimNext(java.util.Set queues) { + long now = now(); + JobRec best = null; + for (JobRec job : jobs.values()) { + if (!"pending".equals(job.status) || job.scheduledAt > now || paused.contains(job.queue)) { + continue; + } + // Honor the worker's queue filter; empty filter = every queue. + if (!queues.isEmpty() && !queues.contains(job.queue)) { + continue; + } + if (best == null || job.priority > best.priority) { + best = job; + } + } + if (best != null) { + best.status = "running"; + best.startedAt = now; + } + return best; + } + + private synchronized void onComplete(JobRec job, byte[] result) { + job.result = result; + job.status = "complete"; + job.completedAt = now(); + } + + private synchronized void onFail(JobRec job, String error) { + if (job.retryCount < job.maxRetries) { + job.retryCount++; + job.status = "pending"; + job.startedAt = null; + job.scheduledAt = now(); + } else { + job.status = "dead"; + job.error = error; + job.completedAt = now(); + Map entry = new LinkedHashMap<>(); + entry.put("id", "dead-" + seq.incrementAndGet()); + entry.put("originalJobId", job.id); + entry.put("queue", job.queue); + entry.put("taskName", job.taskName); + entry.put("error", error); + entry.put("retryCount", job.retryCount); + entry.put("failedAt", now()); + entry.put("metadata", job.metadata); + entry.put("dlqRetryCount", 0); + dead.add(entry); + } + } + + private synchronized void onCancel(JobRec job) { + job.status = "cancelled"; + job.completedAt = now(); + } + + private Map statsFor(String queue) { + Map out = new LinkedHashMap<>(); + out.put("pending", count("pending", queue)); + out.put("running", count("running", queue)); + out.put("completed", count("complete", queue)); + out.put("failed", count("failed", queue)); + out.put("dead", count("dead", queue)); + out.put("cancelled", count("cancelled", queue)); + return out; + } + + private long count(String status, String queue) { + return jobs.values().stream() + .filter(j -> status.equals(j.status) && (queue == null || queue.equals(j.queue))) + .count(); + } + + private static Map jobView(JobRec job) { + Map view = new LinkedHashMap<>(); + view.put("id", job.id); + view.put("queue", job.queue); + view.put("taskName", job.taskName); + view.put("status", job.status); + view.put("priority", job.priority); + view.put("createdAt", job.createdAt); + view.put("scheduledAt", job.scheduledAt); + view.put("startedAt", job.startedAt); + view.put("completedAt", job.completedAt); + view.put("retryCount", job.retryCount); + view.put("maxRetries", job.maxRetries); + view.put("timeoutMs", job.timeoutMs); + view.put("progress", job.progress); + view.put("error", job.error); + view.put("uniqueKey", job.uniqueKey); + view.put("namespace", job.namespace); + view.put("metadata", job.metadata); + return view; + } + + private static String toJson(Object value) { + try { + return JSON.writeValueAsString(value); + } catch (Exception e) { + throw new TaskitoException("failed to encode in-memory view", e); + } + } + + private static JsonNode readNode(String json) { + if (json == null || json.isEmpty()) { + return JSON.nullNode(); + } + try { + return JSON.readTree(json); + } catch (Exception e) { + throw new TaskitoException("failed to read in-memory options", e); + } + } + + private static String text(JsonNode node, String field) { + JsonNode value = node == null ? null : node.get(field); + return value == null || value.isNull() ? null : value.asText(); + } + + private static String optText(JsonNode node, String field, String fallback) { + String value = text(node, field); + return value == null ? fallback : value; + } + + private static int optInt(JsonNode node, String field, int fallback) { + JsonNode value = node == null ? null : node.get(field); + return value == null || value.isNull() ? fallback : value.asInt(); + } + + private static long optLong(JsonNode node, String field, long fallback) { + JsonNode value = node == null ? null : node.get(field); + return value == null || value.isNull() ? fallback : value.asLong(); + } + + /** A stored job. */ + private static final class JobRec { + String id; + String queue = DEFAULT_QUEUE; + String taskName; + byte[] payload; + byte[] result; + volatile String status = "pending"; + int priority; + long createdAt; + long scheduledAt; + Long startedAt; + Long completedAt; + int retryCount; + int maxRetries; + long timeoutMs; + Integer progress; + String error; + String uniqueKey; + String namespace; + String metadata; + volatile boolean cancelRequested; + } + + private static final class LockRec { + final String name; + final String ownerId; + final long acquiredAt; + long expiresAt; + + LockRec(String name, String ownerId, long acquiredAt, long expiresAt) { + this.name = name; + this.ownerId = ownerId; + this.acquiredAt = acquiredAt; + this.expiresAt = expiresAt; + } + } + + /** Polls for dispatchable jobs and drives outcomes back through the bridge. */ + private final class InMemoryWorker implements WorkerControl { + private final WorkerBridge bridge; + private final java.util.Set queues; + private final Map inFlight = new ConcurrentHashMap<>(); + private volatile boolean running = true; + private Thread loop; + + InMemoryWorker(WorkerBridge bridge, java.util.Set queues) { + this.bridge = bridge; + this.queues = queues; + } + + void start() { + loop = new Thread(this::run, "taskito-inmemory-worker"); + loop.setDaemon(true); + loop.start(); + } + + private void run() { + while (running) { + JobRec job; + while (running && (job = claimNext(queues)) != null) { + long token = seq.incrementAndGet(); + inFlight.put(token, job.id); + bridge.onJob(token, job.id, job.taskName, job.payload); + } + sleep(); + } + } + + private void sleep() { + try { + Thread.sleep(5); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + running = false; + } + } + + @Override + public void completeJob(long token, byte[] result) { + String jobId = inFlight.remove(token); + JobRec job = jobs.get(jobId); + if (job == null) { + return; + } + onComplete(job, result); + bridge.onOutcome("success", jobId, job.taskName, null, job.retryCount, false); + } + + @Override + public void failJob(long token, String error) { + String jobId = inFlight.remove(token); + JobRec job = jobs.get(jobId); + if (job == null) { + return; + } + boolean willRetry = job.retryCount < job.maxRetries; + onFail(job, error); + bridge.onOutcome(willRetry ? "retry" : "dead", jobId, job.taskName, error, job.retryCount, false); + } + + @Override + public void cancelJob(long token) { + String jobId = inFlight.remove(token); + JobRec job = jobs.get(jobId); + if (job == null) { + return; + } + onCancel(job); + bridge.onOutcome("cancelled", jobId, job.taskName, null, job.retryCount, false); + } + + @Override + public void stop() { + running = false; + if (loop != null) { + loop.interrupt(); + } + } + + @Override + public void close() { + stop(); + } + } +} diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryTaskito.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryTaskito.java new file mode 100644 index 00000000..a8035d80 --- /dev/null +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryTaskito.java @@ -0,0 +1,24 @@ +package org.byteveda.taskito.test; + +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.serialization.Serializer; + +/** + * Opens a {@link Queue} backed by an {@link InMemoryQueueBackend} — no JNI, no + * disk. Intended for fast unit tests of producers, handlers, retries, and + * dead-lettering. Workflows are not supported in-memory. + */ +public final class InMemoryTaskito { + private InMemoryTaskito() {} + + /** A queue over a fresh in-memory backend using the default JSON serializer. */ + public static Queue open() { + return Taskito.builder().open(new InMemoryQueueBackend()); + } + + /** A queue over a fresh in-memory backend with a custom serializer. */ + public static Queue open(Serializer serializer) { + return Taskito.builder().serializer(serializer).open(new InMemoryQueueBackend()); + } +} diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/package-info.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/package-info.java new file mode 100644 index 00000000..4213c65e --- /dev/null +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/package-info.java @@ -0,0 +1,2 @@ +/** Test support: a pure-Java in-memory {@code QueueBackend} and the {@code InMemoryTaskito} entry point. */ +package org.byteveda.taskito.test; diff --git a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java new file mode 100644 index 00000000..4df52f29 --- /dev/null +++ b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java @@ -0,0 +1,49 @@ +package org.byteveda.taskito.test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobStatus; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +class InMemoryQueueBackendTest { + + @Test + @Timeout(20) + void runsJobToCompletion() throws Exception { + Task dbl = Task.of("im.double", Integer.class); + try (Queue queue = InMemoryTaskito.open()) { // no JNI, no disk + String id = queue.enqueue(dbl, 21); + try (Worker worker = queue.worker().handle(dbl, p -> p * 2).start()) { + Job job = queue.awaitJob(id, Duration.ofSeconds(10)).orElseThrow(); + assertEquals(JobStatus.COMPLETE, job.status); + assertEquals(42, queue.getResult(id, Integer.class).orElseThrow()); + } + } + } + + @Test + @Timeout(20) + void retriesThenDeadLetters() throws Exception { + Task boom = Task.of("im.boom", Integer.class).retries(2); + try (Queue queue = InMemoryTaskito.open()) { + String id = queue.enqueue(boom, 1); + try (Worker worker = queue.worker() + .handle(boom, p -> { + throw new IllegalStateException("boom"); + }) + .start()) { + Job job = queue.awaitJob(id, Duration.ofSeconds(10)).orElseThrow(); + assertEquals(JobStatus.DEAD, job.status); + assertEquals(2, job.retryCount); + assertTrue(queue.listDead(10, 0).size() >= 1); + } + } + } +} diff --git a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/MiddlewareContextTest.java b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/MiddlewareContextTest.java new file mode 100644 index 00000000..1ecf2e60 --- /dev/null +++ b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/MiddlewareContextTest.java @@ -0,0 +1,49 @@ +package org.byteveda.taskito.test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.middleware.EnqueueContext; +import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.middleware.TaskContext; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +class MiddlewareContextTest { + + @Test + @Timeout(20) + void metadataTravelsAndAttributesAreShared() throws Exception { + AtomicReference afterSaw = new AtomicReference<>(); + Task echo = Task.of("mw.echo", Integer.class); + try (Queue queue = InMemoryTaskito.open()) { + queue.use(new Middleware() { + @Override + public void onEnqueue(EnqueueContext ctx) { + ctx.metadata().put("trace-id", "trace-123"); // injected at enqueue + } + + @Override + public void before(TaskContext ctx) { + Object traceId = ctx.job().metadata().get("trace-id"); // read at execution + ctx.attributes().put("seen", traceId); // scratch shared with after() + } + + @Override + public void after(TaskContext ctx, Object result) { + afterSaw.set((String) ctx.attributes().get("seen")); + } + }); + + String id = queue.enqueue(echo, 7); + try (Worker worker = queue.worker().handle(echo, p -> p).start()) { + queue.awaitJob(id, Duration.ofSeconds(10)); + } + assertEquals("trace-123", afterSaw.get()); + } + } +}