From 8ed3ac7e03723bbe460cc6da017703efb71aa4ed Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:23:27 +0530 Subject: [PATCH 1/2] feat(java): enqueue decision gates and recipes --- .../org/byteveda/taskito/DefaultTaskito.java | 99 +++++++++++++--- .../java/org/byteveda/taskito/Taskito.java | 21 ++++ .../errors/EnqueueSkippedException.java | 14 +++ .../taskito/predicates/BooleanFlag.java | 7 ++ .../taskito/predicates/EnqueueDecision.java | 57 +++++++++ .../taskito/predicates/EnqueueGate.java | 12 ++ .../byteveda/taskito/predicates/Recipes.java | 109 ++++++++++++++++++ 7 files changed, 300 insertions(+), 19 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/errors/EnqueueSkippedException.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/predicates/BooleanFlag.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueDecision.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueGate.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/predicates/Recipes.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index c73e5439..94a02532 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -17,6 +17,7 @@ import java.util.function.Consumer; import java.util.function.Function; import org.byteveda.taskito.core.CoreFacade; +import org.byteveda.taskito.errors.EnqueueSkippedException; import org.byteveda.taskito.errors.InterceptionException; import org.byteveda.taskito.errors.PredicateRejectedException; import org.byteveda.taskito.errors.SerializationException; @@ -36,6 +37,8 @@ import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.predicates.EnqueueDecision; +import org.byteveda.taskito.predicates.EnqueueGate; import org.byteveda.taskito.predicates.Predicate; import org.byteveda.taskito.predicates.PredicateContext; import org.byteveda.taskito.resources.ResourceContext; @@ -72,7 +75,7 @@ final class DefaultTaskito implements Taskito { private final Map codecs; private final List middleware = new CopyOnWriteArrayList<>(); private final ResourceRuntime resources = new ResourceRuntime(); - private final Map> predicates = new ConcurrentHashMap<>(); + private final Map> gates = new ConcurrentHashMap<>(); private final List interceptors = new CopyOnWriteArrayList<>(); DefaultTaskito(QueueBackend backend, Serializer serializer, Map codecs) { @@ -120,9 +123,16 @@ public Map resourceMetrics() { @Override public Taskito predicate(String taskName, Predicate predicate) { - predicates - .computeIfAbsent(taskName, key -> new CopyOnWriteArrayList<>()) - .add(predicate); + return gate( + taskName, + context -> predicate.test(context) + ? EnqueueDecision.allow() + : EnqueueDecision.reject("enqueue of task '" + taskName + "' rejected by a predicate")); + } + + @Override + public Taskito gate(String taskName, EnqueueGate gate) { + gates.computeIfAbsent(taskName, key -> new CopyOnWriteArrayList<>()).add(gate); return this; } @@ -146,16 +156,42 @@ public String enqueue(Task task, T payload) { @Override public String enqueue(Task task, T payload, EnqueueOptions options) { - return dispatchEnqueue(task.name(), payload, options, task.codecNames()); + return required(dispatchEnqueue(task.name(), payload, options, task.codecNames()), task.name()); } @Override public String enqueue(String taskName, Object payload) { + return required(dispatchEnqueue(taskName, payload, EnqueueOptions.none(), List.of()), taskName); + } + + @Override + public Optional tryEnqueue(Task task, T payload) { + return tryEnqueue(task, payload, task.options()); + } + + @Override + public Optional tryEnqueue(Task task, T payload, EnqueueOptions options) { + return dispatchEnqueue(task.name(), payload, options, task.codecNames()); + } + + @Override + public Optional tryEnqueue(String taskName, Object payload) { return dispatchEnqueue(taskName, payload, EnqueueOptions.none(), List.of()); } - /** Interceptors + onEnqueue middleware, then serialize, codec-encode, and submit the job. */ - private String dispatchEnqueue(String taskName, Object payload, EnqueueOptions options, List codecNames) { + /** Unwrap a dispatch result, turning a gate {@code Skip} into an {@link EnqueueSkippedException}. */ + private static String required(Optional jobId, String taskName) { + return jobId.orElseThrow(() -> + new EnqueueSkippedException("enqueue of '" + taskName + "' was skipped by a gate; use tryEnqueue")); + } + + /** + * Interceptors → onEnqueue middleware → enqueue gates, then serialize, + * codec-encode, and submit. Returns the job id, or empty when a gate skips + * the enqueue; throws when a gate rejects it. + */ + private Optional dispatchEnqueue( + String taskName, Object payload, EnqueueOptions options, List codecNames) { String originalTaskName = taskName; for (Interceptor interceptor : interceptors) { Interception outcome = interceptor.intercept(taskName, payload); @@ -185,16 +221,31 @@ private String dispatchEnqueue(String taskName, Object payload, EnqueueOptions o m.onEnqueue(context); } // Gate the payload that will actually be enqueued (after interceptors and - // middleware may have rewritten it). - gate(taskName, context.payload()); + // middleware may have rewritten it); the first non-Allow decision wins. + EnqueueDecision decision = evaluate(taskName, context.payload()); + if (decision instanceof EnqueueDecision.Reject reject) { + throw new PredicateRejectedException("enqueue of '" + taskName + "' rejected: " + reject.reason()); + } + if (decision instanceof EnqueueDecision.Skip) { + return Optional.empty(); + } EnqueueOptions finalOptions = context.options(); + if (decision instanceof EnqueueDecision.Defer defer) { + finalOptions = withDelay(finalOptions, defer.delay()); + } if (!context.metadata().isEmpty()) { finalOptions = finalOptions.toBuilder() .metadata(encode(context.metadata())) .build(); } byte[] data = encodeCodecs(serializer.serialize(context.payload()), codecNames); - return backend.enqueue(taskName, data, encode(finalOptions)); + return Optional.of(backend.enqueue(taskName, data, encode(finalOptions))); + } + + /** Apply a defer delay to the options (overriding any delay already set). */ + private static EnqueueOptions withDelay(EnqueueOptions options, Duration delay) { + long delayMs = delay.toMillis(); + return delayMs <= 0 ? options : options.toBuilder().delayMs(delayMs).build(); } /** Apply a task's payload codecs in order; throws if a name is not registered. */ @@ -210,18 +261,20 @@ private byte[] encodeCodecs(byte[] data, List codecNames) { return bytes; } - /** Run any registered enqueue gates for {@code taskName}; throw if one rejects. */ - private void gate(String taskName, Object payload) { - List gates = predicates.get(taskName); - if (gates == null || gates.isEmpty()) { - return; + /** Evaluate registered gates in order; the first non-{@code Allow} decision wins. */ + private EnqueueDecision evaluate(String taskName, Object payload) { + List taskGates = gates.get(taskName); + if (taskGates == null || taskGates.isEmpty()) { + return EnqueueDecision.allow(); } PredicateContext context = new PredicateContext(taskName, payload); - for (Predicate gate : gates) { - if (!gate.test(context)) { - throw new PredicateRejectedException("enqueue of task '" + taskName + "' rejected by a predicate"); + for (EnqueueGate gate : taskGates) { + EnqueueDecision decision = gate.decide(context); + if (!(decision instanceof EnqueueDecision.Allow)) { + return decision; } } + return EnqueueDecision.allow(); } @Override @@ -238,7 +291,15 @@ public List enqueueMany(Task task, List payloads, EnqueueOptio List perJob = new ArrayList<>(payloads.size()); for (int i = 0; i < payloads.size(); i++) { Object payload = interceptBatchPayload(task.name(), payloads.get(i)); - gate(task.name(), payload); + EnqueueDecision decision = evaluate(task.name(), payload); + if (decision instanceof EnqueueDecision.Reject reject) { + throw new PredicateRejectedException("enqueue of '" + task.name() + "' rejected: " + reject.reason()); + } + if (!(decision instanceof EnqueueDecision.Allow)) { + // Skip/Defer can't be expressed per-item in a single all-or-nothing batch. + throw new EnqueueSkippedException("enqueue of '" + task.name() + + "' was skipped or deferred by a gate; batch enqueue needs Allow"); + } bytes[i] = encodeCodecs(serializer.serialize(payload), task.codecNames()); perJob.add(options); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index 589cd529..95dc280f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -30,6 +30,7 @@ import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.predicates.EnqueueGate; import org.byteveda.taskito.predicates.Predicate; import org.byteveda.taskito.resources.ResourceContext; import org.byteveda.taskito.resources.ResourceScope; @@ -87,6 +88,14 @@ static Builder builder() { */ Taskito predicate(String taskName, Predicate predicate); + /** + * Gate enqueues of {@code taskName} with a richer {@link EnqueueGate} whose + * {@link org.byteveda.taskito.predicates.EnqueueDecision} can allow, skip, + * defer, or reject. Gates run in registration order and the first non-allow + * decision wins. Returns {@code this}. + */ + Taskito gate(String taskName, EnqueueGate gate); + /** * Register an interceptor that may convert, redirect, or reject each enqueue * before it is serialized (see {@link Interceptor}). Returns {@code this}. @@ -103,6 +112,18 @@ static Builder builder() { /** Enqueue by task name with an arbitrary payload and default options. */ String enqueue(String taskName, Object payload); + /** + * Like {@link #enqueue(Task, Object)} but gate-aware: returns the job id, or + * an empty {@code Optional} when a gate skips the enqueue. A gate + * {@code Reject} still throws. + */ + Optional tryEnqueue(Task task, T payload); + + Optional tryEnqueue(Task task, T payload, EnqueueOptions options); + + /** Gate-aware {@link #enqueue(String, Object)}; empty when a gate skips the enqueue. */ + Optional tryEnqueue(String taskName, Object payload); + /** Enqueue a batch in one storage call; returns ids in input order. */ List enqueueMany(Task task, List payloads); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/EnqueueSkippedException.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/EnqueueSkippedException.java new file mode 100644 index 00000000..71807536 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/EnqueueSkippedException.java @@ -0,0 +1,14 @@ +package org.byteveda.taskito.errors; + +import org.byteveda.taskito.TaskitoException; + +/** + * A gate returned {@code Skip} for an enqueue, so no job was created. Thrown by + * {@code enqueue}; callers that expect skips should use {@code tryEnqueue}, + * which returns an empty {@code Optional} instead. + */ +public class EnqueueSkippedException extends TaskitoException { + public EnqueueSkippedException(String message) { + super(message); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/predicates/BooleanFlag.java b/sdks/java/src/main/java/org/byteveda/taskito/predicates/BooleanFlag.java new file mode 100644 index 00000000..f0b35746 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/predicates/BooleanFlag.java @@ -0,0 +1,7 @@ +package org.byteveda.taskito.predicates; + +/** A feature-flag lookup seam, plugged into {@link Recipes#featureFlag}. */ +@FunctionalInterface +public interface BooleanFlag { + boolean enabled(String flag); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueDecision.java b/sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueDecision.java new file mode 100644 index 00000000..fa770a6e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueDecision.java @@ -0,0 +1,57 @@ +package org.byteveda.taskito.predicates; + +import java.time.Duration; +import java.time.Instant; + +/** + * The outcome of an {@link EnqueueGate}: allow the enqueue, silently skip it, + * defer it by a delay, or reject it with a reason. Build with the static + * factories and pattern-match the variants where decisions are honored. + */ +public sealed interface EnqueueDecision + permits EnqueueDecision.Allow, EnqueueDecision.Skip, EnqueueDecision.Defer, EnqueueDecision.Reject { + + /** Proceed with the enqueue unchanged. */ + record Allow() implements EnqueueDecision {} + + /** Do not enqueue: {@code tryEnqueue} returns empty, {@code enqueue} throws. */ + record Skip(String reason) implements EnqueueDecision {} + + /** Enqueue, but delayed by {@code delay} (overrides any delay in the passed options). */ + record Defer(Duration delay) implements EnqueueDecision { + public Defer { + if (delay == null || delay.isNegative()) { + throw new IllegalArgumentException("defer delay must be non-negative"); + } + } + } + + /** Refuse the enqueue: both {@code enqueue} and {@code tryEnqueue} throw with {@code reason}. */ + record Reject(String reason) implements EnqueueDecision {} + + static EnqueueDecision allow() { + return new Allow(); + } + + static EnqueueDecision skip() { + return new Skip(""); + } + + static EnqueueDecision skip(String reason) { + return new Skip(reason == null ? "" : reason); + } + + static EnqueueDecision defer(Duration delay) { + return new Defer(delay); + } + + /** Defer until {@code instant}; resolves to a delay from now (zero if already past). */ + static EnqueueDecision deferUntil(Instant instant) { + Duration delay = Duration.between(Instant.now(), instant); + return new Defer(delay.isNegative() ? Duration.ZERO : delay); + } + + static EnqueueDecision reject(String reason) { + return new Reject(reason == null ? "" : reason); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueGate.java b/sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueGate.java new file mode 100644 index 00000000..c2876dc4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/predicates/EnqueueGate.java @@ -0,0 +1,12 @@ +package org.byteveda.taskito.predicates; + +/** + * A richer enqueue gate that maps a {@link PredicateContext} to an + * {@link EnqueueDecision} (allow / skip / defer / reject). Evaluated + * synchronously on the producer thread, so keep it fast and side-effect-free. + * For a plain boolean allow/reject, use {@link Predicate} instead. + */ +@FunctionalInterface +public interface EnqueueGate { + EnqueueDecision decide(PredicateContext context); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/predicates/Recipes.java b/sdks/java/src/main/java/org/byteveda/taskito/predicates/Recipes.java new file mode 100644 index 00000000..feb32f57 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/predicates/Recipes.java @@ -0,0 +1,109 @@ +package org.byteveda.taskito.predicates; + +import java.time.DayOfWeek; +import java.time.Duration; +import java.time.LocalDate; +import java.time.LocalTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.Set; + +/** + * Ready-made {@link EnqueueGate}s for common scheduling and filtering policies. + * Time-based recipes {@code defer} out-of-window enqueues to the next open + * moment; filter recipes {@code skip} non-matching payloads. + */ +public final class Recipes { + private Recipes() {} + + private static final LocalTime DEFAULT_OPEN = LocalTime.of(9, 0); + private static final LocalTime DEFAULT_CLOSE = LocalTime.of(17, 0); + private static final Set WEEKDAYS = + EnumSet.of(DayOfWeek.MONDAY, DayOfWeek.TUESDAY, DayOfWeek.WEDNESDAY, DayOfWeek.THURSDAY, DayOfWeek.FRIDAY); + + /** Allow weekday 09:00–17:00 in {@code zone}; otherwise defer to the next opening. */ + public static EnqueueGate businessHours(ZoneId zone) { + return businessHours(zone, DEFAULT_OPEN, DEFAULT_CLOSE); + } + + /** Allow Mon–Fri within {@code [open, close)} in {@code zone}; otherwise defer to the next opening. */ + public static EnqueueGate businessHours(ZoneId zone, LocalTime open, LocalTime close) { + return context -> { + ZonedDateTime now = ZonedDateTime.now(zone); + LocalTime time = now.toLocalTime(); + boolean openNow = WEEKDAYS.contains(now.getDayOfWeek()) && !time.isBefore(open) && time.isBefore(close); + if (openNow) { + return EnqueueDecision.allow(); + } + return EnqueueDecision.defer(Duration.between(now, nextOpening(now, open))); + }; + } + + /** Allow within a daily {@code [start, end)} window in {@code zone} (wraps past midnight); else defer to start. */ + public static EnqueueGate timeWindow(ZoneId zone, LocalTime start, LocalTime end) { + return context -> { + ZonedDateTime now = ZonedDateTime.now(zone); + LocalTime time = now.toLocalTime(); + boolean inWindow = start.isBefore(end) + ? !time.isBefore(start) && time.isBefore(end) + : !time.isBefore(start) || time.isBefore(end); + if (inWindow) { + return EnqueueDecision.allow(); + } + ZonedDateTime target = now.toLocalDate().atTime(start).atZone(zone); + if (!target.isAfter(now)) { + target = target.plusDays(1); + } + return EnqueueDecision.defer(Duration.between(now, target)); + }; + } + + /** Allow on the given days in {@code zone}; otherwise defer to the start of the next allowed day. */ + public static EnqueueGate dayOfWeek(ZoneId zone, DayOfWeek... allowed) { + Set days = + allowed.length == 0 ? EnumSet.noneOf(DayOfWeek.class) : EnumSet.copyOf(Arrays.asList(allowed)); + return context -> { + ZonedDateTime now = ZonedDateTime.now(zone); + if (days.contains(now.getDayOfWeek())) { + return EnqueueDecision.allow(); + } + LocalDate date = now.toLocalDate().plusDays(1); + for (int i = 0; i < 7; i++) { + if (days.contains(date.getDayOfWeek())) { + return EnqueueDecision.defer(Duration.between(now, date.atStartOfDay(zone))); + } + date = date.plusDays(1); + } + return EnqueueDecision.skip("no allowed day of week"); + }; + } + + /** Allow only when {@code test} accepts the payload; otherwise skip silently. */ + public static EnqueueGate payloadMatches(java.util.function.Predicate test) { + return context -> + test.test(context.payload()) ? EnqueueDecision.allow() : EnqueueDecision.skip("payload did not match"); + } + + /** Allow only when {@code flag} is enabled; otherwise skip silently. */ + public static EnqueueGate featureFlag(String flag, BooleanFlag flags) { + return context -> + flags.enabled(flag) ? EnqueueDecision.allow() : EnqueueDecision.skip("feature '" + flag + "' disabled"); + } + + /** The next weekday opening at or after {@code now} (today at {@code open} if still ahead). */ + private static ZonedDateTime nextOpening(ZonedDateTime now, LocalTime open) { + if (WEEKDAYS.contains(now.getDayOfWeek()) && now.toLocalTime().isBefore(open)) { + return now.toLocalDate().atTime(open).atZone(now.getZone()); + } + LocalDate date = now.toLocalDate().plusDays(1); + for (int i = 0; i < 7; i++) { + if (WEEKDAYS.contains(date.getDayOfWeek())) { + return date.atTime(open).atZone(now.getZone()); + } + date = date.plusDays(1); + } + return now; + } +} From abe33a58299b35e3e6909666090673cd047a5889 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:23:27 +0530 Subject: [PATCH 2/2] test(java): cover enqueue gates and recipes --- .../byteveda/taskito/EnqueueDecisionTest.java | 89 +++++++++++++++++++ .../org/byteveda/taskito/RecipesTest.java | 69 ++++++++++++++ 2 files changed, 158 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/RecipesTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java new file mode 100644 index 00000000..c0e0d53c --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java @@ -0,0 +1,89 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Optional; +import org.byteveda.taskito.errors.EnqueueSkippedException; +import org.byteveda.taskito.errors.PredicateRejectedException; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.predicates.EnqueueDecision; +import org.byteveda.taskito.task.Task; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class EnqueueDecisionTest { + + private static final Task TASK = Task.of("ed.echo", String.class); + + @Test + void skipReturnsEmptyFromTryEnqueueAndThrowsFromEnqueue(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "skip")) { + queue.gate(TASK.name(), context -> EnqueueDecision.skip("not now")); + + assertEquals(Optional.empty(), queue.tryEnqueue(TASK, "hi")); + assertThrows(EnqueueSkippedException.class, () -> queue.enqueue(TASK, "hi")); + } + } + + @Test + void rejectThrowsFromBothPaths(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "reject")) { + queue.gate(TASK.name(), context -> EnqueueDecision.reject("nope")); + + assertThrows(PredicateRejectedException.class, () -> queue.enqueue(TASK, "hi")); + assertThrows(PredicateRejectedException.class, () -> queue.tryEnqueue(TASK, "hi")); + } + } + + @Test + void deferDelaysTheScheduledTime(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "defer")) { + queue.gate(TASK.name(), context -> EnqueueDecision.defer(Duration.ofHours(1))); + + Optional id = queue.tryEnqueue(TASK, "later"); + assertTrue(id.isPresent()); + Job job = queue.getJob(id.get()).orElseThrow(); + long delayMs = job.scheduledAt - job.createdAt; + assertTrue(delayMs >= Duration.ofMinutes(59).toMillis(), "expected ~1h defer, got " + delayMs + "ms"); + } + } + + @Test + void allowEnqueuesNormally(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "allow")) { + queue.gate(TASK.name(), context -> EnqueueDecision.allow()); + + String id = queue.enqueue(TASK, "hi"); + assertTrue(queue.getJob(id).isPresent()); + } + } + + @Test + void booleanPredicateStillRejects(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "pred")) { + queue.predicate(TASK.name(), context -> false); + + assertThrows(PredicateRejectedException.class, () -> queue.enqueue(TASK, "hi")); + } + } + + @Test + void firstNonAllowGateWins(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "order")) { + queue.gate(TASK.name(), context -> EnqueueDecision.skip("first")); + queue.gate(TASK.name(), context -> EnqueueDecision.reject("second")); + + // Skip is registered first, so the enqueue is skipped (not rejected). + assertFalse(queue.tryEnqueue(TASK, "hi").isPresent()); + } + } + + private static Taskito open(Path dir, String name) { + return Taskito.builder().url(dir.resolve(name + ".db").toString()).open(); + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/RecipesTest.java b/sdks/java/src/test/java/org/byteveda/taskito/RecipesTest.java new file mode 100644 index 00000000..1d7040f3 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/RecipesTest.java @@ -0,0 +1,69 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.time.LocalTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import org.byteveda.taskito.predicates.EnqueueDecision; +import org.byteveda.taskito.predicates.EnqueueGate; +import org.byteveda.taskito.predicates.PredicateContext; +import org.byteveda.taskito.predicates.Recipes; +import org.junit.jupiter.api.Test; + +class RecipesTest { + + private static final ZoneId UTC = ZoneId.of("UTC"); + private static final PredicateContext CTX = new PredicateContext("t", "payload"); + + @Test + void payloadMatchesAllowsAndSkips() { + EnqueueGate gate = Recipes.payloadMatches(payload -> "yes".equals(payload)); + + assertInstanceOf(EnqueueDecision.Allow.class, gate.decide(new PredicateContext("t", "yes"))); + assertInstanceOf(EnqueueDecision.Skip.class, gate.decide(new PredicateContext("t", "no"))); + } + + @Test + void featureFlagAllowsWhenEnabledSkipsOtherwise() { + assertInstanceOf( + EnqueueDecision.Allow.class, + Recipes.featureFlag("beta", flag -> true).decide(CTX)); + assertInstanceOf( + EnqueueDecision.Skip.class, + Recipes.featureFlag("beta", flag -> false).decide(CTX)); + } + + @Test + void dayOfWeekAllowsToday() { + var today = ZonedDateTime.now(UTC).getDayOfWeek(); + assertInstanceOf( + EnqueueDecision.Allow.class, Recipes.dayOfWeek(UTC, today).decide(CTX)); + } + + @Test + void dayOfWeekDefersToOtherDay() { + var tomorrow = ZonedDateTime.now(UTC).plusDays(1).getDayOfWeek(); + EnqueueDecision decision = Recipes.dayOfWeek(UTC, tomorrow).decide(CTX); + EnqueueDecision.Defer defer = assertInstanceOf(EnqueueDecision.Defer.class, decision); + assertTrue(!defer.delay().isNegative() && defer.delay().compareTo(Duration.ofDays(2)) <= 0); + } + + @Test + void timeWindowAllowsWhenWindowSpansTheDay() { + EnqueueGate gate = Recipes.timeWindow(UTC, LocalTime.MIN, LocalTime.MAX); + assertInstanceOf(EnqueueDecision.Allow.class, gate.decide(CTX)); + } + + @Test + void businessHoursAllowsOrDefersWithinAWeek() { + EnqueueDecision decision = Recipes.businessHours(UTC).decide(CTX); + if (decision instanceof EnqueueDecision.Defer defer) { + assertTrue(defer.delay().compareTo(Duration.ofDays(7)) <= 0); + } else { + assertInstanceOf(EnqueueDecision.Allow.class, decision); + } + } +}