From a8fa2ed09a8f210c3dda39a249f297e7fce49aff Mon Sep 17 00:00:00 2001 From: Brandur Date: Tue, 27 Oct 2026 15:21:29 -0500 Subject: [PATCH] [java] Add `QUEUE_DEFAULT` constant like other languages I just happened to notice while looking at Java docs samples that while all other languages have some sort of constant representing the default queue, Java never got one for some reason, and so its code samples all use a literal string like `"default"`. | Language | Public constant | |---|---| | Go | `river.QueueDefault` | | JavaScript / TypeScript | `QUEUE_DEFAULT` from `"riverqueue"` | | Ruby | `River::QUEUE_DEFAULT` | | Rust | `riverqueue::QUEUE_DEFAULT` | Here, add `Client.QUEUE_DEFAULT` to fill the gap. Has no functional effect, but makes cross-language packages a little more consistent, and makes the code samples look a little nicer IMO. --- java/API.md | 1 + java/CHANGELOG.md | 4 ++++ java/README.md | 21 +++++++++++-------- .../conformance/RuntimeAdapter.java | 4 +++- .../src/main/java/com/riverqueue/Client.java | 3 +++ .../java/com/riverqueue/InsertOptions.java | 3 ++- .../com/riverqueue/ClientLifecycleTest.java | 4 ++-- 7 files changed, 27 insertions(+), 13 deletions(-) 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(