diff --git a/java/API.md b/java/API.md index 92a6ae2ce..89681a021 100644 --- a/java/API.md +++ b/java/API.md @@ -19,6 +19,7 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G | `ResumableSetStepTx`, `ResumableSetStepCursorTx` | `WorkContext.checkpoint(connection[, cursor])` | Checkpoints commit with application writes; failed transactions leave the last committed progress intact. | | Worker timeout | `Workers.Handler.timeout(job)` | A default method preserves lambda handlers; per-job overrides also govern rescue decisions. | | Retry policy | `RetryPolicy` | A lambda receives the full job and returns `Duration`; the default uses the same error-count backoff and jitter. | +| `QueueDefault` | `Client.QUEUE_DEFAULT` | The default queue name, `"default"`, used when neither insert options nor job-type defaults specify a queue. | | `Queues`, queue CRUD/control APIs | `RunHandle.addQueue/removeQueue`, `Client.queues()` | Runtime concurrency and persisted queue state have separate owners. Reads and mutations support caller transactions. | | `Subscribe`, `EventKind`, `Event` | `Client.subscribe`, `EventKind`, `Event`, `Subscription` | Enum filters and an `AutoCloseable` subscription replace channels. Callbacks are synchronous; queue events include their queue. | | Periodic jobs and schedules | `Client.Builder.periodic`, `RunHandle.periodicJobs()`, `PeriodicJobBundle`, `PeriodicJob`, `Schedule` | Java time types and a functional scheduling interface; cron implementation details are package-private. | diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index 4d45e1841..31072b582 100644 --- a/java/CHANGELOG.md +++ b/java/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Expose `Client.QUEUE_DEFAULT` (`"default"`) for queue configuration and operations, matching the default queue constant in other River clients. + ## [0.2.0] - 2026-10-26 ### Changed diff --git a/java/README.md b/java/README.md index f0c3aa511..080d8fb71 100644 --- a/java/README.md +++ b/java/README.md @@ -58,7 +58,7 @@ public class Example { System.out.println("Sending " + context.args().subject() + " to " + context.args().address()); context.output("sent"); }); - var client = Client.builder(database).queue("default", 10).workers(workers).build(); + var client = Client.builder(database).queue(Client.QUEUE_DEFAULT, 10).workers(workers).build(); var completed = new CompletableFuture(); try (var subscription = client.subscribe(completed::complete, Client.EventKind.JOB_COMPLETED); var run = client.start()) { @@ -274,7 +274,7 @@ time, River reports the problem and uses its default backoff. ```java var retryingEmail = email.withDefaults(InsertOptions.builder().maxAttempts(10).build()); var clientBuilder = Client.builder(database) - .queue("default", 10) + .queue(Client.QUEUE_DEFAULT, 10) .retryPolicy(job -> Duration.ofSeconds(Math.min(300, 5L * (job.errors().size() + 1)))) .worker(retryingEmail, context -> mailer.send(context.args().address(), context.args().subject())); @@ -300,7 +300,7 @@ existing job's ID. ```java var email = JobType.of("send_email", SendEmail.class); -var clientBuilder = Client.builder(database).queue("default", 10); +var clientBuilder = Client.builder(database).queue(Client.QUEUE_DEFAULT, 10); ``` ### Cancelling jobs @@ -383,7 +383,7 @@ HookWorkEnd reportFailure = (context, failure) -> { return failure; }; var clientBuilder = Client.builder(database).plugins(reportFailure) - .queue("default", 10) + .queue(Client.QUEUE_DEFAULT, 10) .errorHandler(failure -> System.getLogger("river").log(System.Logger.Level.ERROR, "Runtime failure", failure)) .worker(email, context -> mailer.send(context.args().address(), context.args().subject())); @@ -504,9 +504,12 @@ not capture SLF4J or `System.Logger` output automatically. different workloads. Set a queue at insertion and configure its worker count. The count is local to each runtime; it is not a cluster-wide concurrency limit. +`Client.QUEUE_DEFAULT` is the default queue name, `"default"`. Jobs use it when +neither insertion options nor job-type defaults specify a queue. + ```java var clientBuilder = Client.builder(database) - .queue("default", 10) + .queue(Client.QUEUE_DEFAULT, 10) .queue("mail", 4) .worker(email, context -> mailer.send(context.args().address(), context.args().subject())); client.insert(email, new SendEmail("hello@example.com", "Welcome"), @@ -551,7 +554,7 @@ languages match only when they use the same time-zone data. ```java var clientBuilder = Client.builder(database) - .queue("default", 10) + .queue(Client.QUEUE_DEFAULT, 10) .worker(email, context -> mailer.send(context.args().address(), context.args().subject())) .periodic("hourly-status", Schedule.every(Duration.ofHours(1)), email, new SendEmail("ops@example.com", "Hourly status"), InsertOptions.defaults(), true) @@ -718,7 +721,7 @@ void completes(@org.junit.jupiter.api.io.TempDir java.nio.file.Path directory) t var database = Database.connect("jdbc:sqlite:" + directory.resolve("river.db")); new Migrator(database).migrate(); var client = Client.builder(database) - .queue("default", 1) + .queue(Client.QUEUE_DEFAULT, 1) .worker(echo, context -> context.output(context.args().message())) .build(); var completed = new java.util.concurrent.CompletableFuture(); @@ -850,7 +853,7 @@ consuming an attempt. `awaitCancellation()` only waits for the signal. ```java var client = Client.builder(database) - .queue("default", 10) + .queue(Client.QUEUE_DEFAULT, 10) .stopTimeout(Duration.ofSeconds(30)) .worker(email, context -> mailer.send(context.args().address(), context.args().subject())) .build(); @@ -883,7 +886,7 @@ delays must fit in signed 64-bit nanoseconds, up to about 292 years. ```java var clientBuilder = Client.builder(database) - .queue("default", 10) + .queue(Client.QUEUE_DEFAULT, 10) .retention(Duration.ofDays(1), Duration.ofDays(1), Duration.ofDays(7)) .worker(email, context -> mailer.send(context.args().address(), context.args().subject())); ``` diff --git a/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java b/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java index 2d18408b3..1d6a21742 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java @@ -185,7 +185,9 @@ private void start(Client river, JsonNode params) { var builder = river.toBuilder() .id(params.path("client_id").asString()) - .queue(params.path("queue").asString("default"), params.path("max_workers").asInt(1)) + .queue( + params.path("queue").asString(Client.QUEUE_DEFAULT), + params.path("max_workers").asInt(1)) .worker(ECHO, this::work) .pollOnly(params.path("poll_only").asBoolean(false)) .pollInterval(Duration.ofMillis(params.path("fetch_poll_interval_ms").asLong(10))) diff --git a/java/river/src/main/java/com/riverqueue/Client.java b/java/river/src/main/java/com/riverqueue/Client.java index 752f5a9d8..cff7b2f0b 100644 --- a/java/river/src/main/java/com/riverqueue/Client.java +++ b/java/river/src/main/java/com/riverqueue/Client.java @@ -28,6 +28,9 @@ * application owns its connection pool; inserting jobs does not require a running client. */ public final class Client { + /** Default queue name, used when neither insert options nor job-type defaults specify a queue. */ + public static final String QUEUE_DEFAULT = "default"; + private final Builder config; private final String id; private final List> subscribers = new CopyOnWriteArrayList<>(); diff --git a/java/river/src/main/java/com/riverqueue/InsertOptions.java b/java/river/src/main/java/com/riverqueue/InsertOptions.java index c27a72923..fd1eab640 100644 --- a/java/river/src/main/java/com/riverqueue/InsertOptions.java +++ b/java/river/src/main/java/com/riverqueue/InsertOptions.java @@ -38,7 +38,7 @@ InsertOptions resolve(InsertOptions fallback) { first(metadata, fallback.metadata, Json.object()), first(pending, fallback.pending, false), first(priority, fallback.priority, 1), - first(queue, fallback.queue, "default"), + first(queue, fallback.queue, Client.QUEUE_DEFAULT), first(scheduledAt, fallback.scheduledAt, null), first(tags, fallback.tags, List.of()), first(unique, fallback.unique, Unique.none())); @@ -84,6 +84,7 @@ public Builder priority(int value) { return this; } + /** Sets the queue; defaults to the job type's queue or {@link Client#QUEUE_DEFAULT}. */ public Builder queue(String value) { queue = value; return this; diff --git a/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java b/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java index 6fda7cf5e..49d5ad231 100644 --- a/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java +++ b/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java @@ -32,7 +32,7 @@ private Client.Builder builder() { .leadership(false) .pollInterval(Duration.ofMillis(10)) .pollOnly(true) - .queue("default", 1) + .queue(Client.QUEUE_DEFAULT, 1) .stopTimeout(Duration.ofSeconds(5)); } @@ -118,7 +118,7 @@ void failedDatabaseStartupCanBeRetried() throws Exception { Client.builder(new Database(source, database.dialect()).withSchema(database.schema())) .leadership(false) .pollOnly(true) - .queue("default", 1) + .queue(Client.QUEUE_DEFAULT, 1) .workers(new Workers().add(TYPE, context -> {})) .build(); assertEquals(