From f7bafb0d5224dd3c7a855ffe7963f2474b89e376 Mon Sep 17 00:00:00 2001 From: Brandur Date: Sat, 17 Oct 2026 22:47:24 -0500 Subject: [PATCH] Java: Standardize hooks + middleware implementation with other languages I noticed today that Java had been done something a little weird in that instead of hooks/middleware like in Go, we ended up with a broad `Extension` interface that does everything. Here, give Java a more normal implementation compared to other languages. We have specific hooks/middleware like: public interface HookInsertBegin extends Hook { /** * Runs in the insertion's transaction. Throwing fails and rolls back the insertion, including * application writes made using this connection. Options are the caller's overrides, before * merging with job-type defaults. */ void insertBegin(Connection connection, JobType type, Object args, InsertOptions options) throws Exception; } Which extends from a base like `Hook`: public interface Hook extends Plugin {} Which extends from `Plugin` so that everything can be set together: public interface Plugin {} --- java/API.md | 12 +- java/README.md | 101 +++++- java/conformance/feature-inventory.json | 144 ++++---- .../com/riverqueue/conformance/Adapter.java | 7 +- .../conformance/RuntimeAdapter.java | 96 +++--- .../src/main/java/com/riverqueue/Client.java | 182 +++++++--- .../main/java/com/riverqueue/Extension.java | 27 -- .../src/main/java/com/riverqueue/Hook.java | 10 + .../java/com/riverqueue/HookInsertBegin.java | 15 + .../com/riverqueue/HookPeriodicJobsStart.java | 11 + .../java/com/riverqueue/HookWorkBegin.java | 11 + .../main/java/com/riverqueue/HookWorkEnd.java | 13 + .../com/riverqueue/JobInsertMiddleware.java | 21 ++ .../main/java/com/riverqueue/Middleware.java | 9 + .../src/main/java/com/riverqueue/Pilot.java | 110 ++++++ .../src/main/java/com/riverqueue/Plugin.java | 115 +------ .../java/com/riverqueue/WorkerMiddleware.java | 11 + .../src/main/java/com/riverqueue/Workers.java | 54 +-- .../java/com/riverqueue/package-info.java | 2 +- .../java/com/riverqueue/BulkDeleteTest.java | 12 +- .../test/java/com/riverqueue/ClientTest.java | 16 +- .../com/riverqueue/HookMiddlewareTest.java | 312 ++++++++++++++++++ .../java/com/riverqueue/LeadershipTest.java | 38 +-- .../riverqueue/NotificationCleanupTest.java | 4 +- .../java/com/riverqueue/NotificationTest.java | 4 +- .../com/riverqueue/PeerCancellationTest.java | 4 +- .../test/java/com/riverqueue/PluginTest.java | 237 +++++++++++++ .../com/riverqueue/ProtocolStorageTest.java | 4 +- .../test/java/com/riverqueue/RescueTest.java | 8 +- .../test/java/com/riverqueue/RetryTest.java | 4 +- .../java/com/riverqueue/SchedulingTest.java | 12 +- .../com/riverqueue/VirtualThreadTest.java | 10 +- .../test/java/com/riverqueue/WorkersTest.java | 34 +- 33 files changed, 1235 insertions(+), 415 deletions(-) delete mode 100644 java/river/src/main/java/com/riverqueue/Extension.java create mode 100644 java/river/src/main/java/com/riverqueue/Hook.java create mode 100644 java/river/src/main/java/com/riverqueue/HookInsertBegin.java create mode 100644 java/river/src/main/java/com/riverqueue/HookPeriodicJobsStart.java create mode 100644 java/river/src/main/java/com/riverqueue/HookWorkBegin.java create mode 100644 java/river/src/main/java/com/riverqueue/HookWorkEnd.java create mode 100644 java/river/src/main/java/com/riverqueue/JobInsertMiddleware.java create mode 100644 java/river/src/main/java/com/riverqueue/Middleware.java create mode 100644 java/river/src/main/java/com/riverqueue/Pilot.java create mode 100644 java/river/src/main/java/com/riverqueue/WorkerMiddleware.java create mode 100644 java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java create mode 100644 java/river/src/test/java/com/riverqueue/PluginTest.java diff --git a/java/API.md b/java/API.md index e047f08eb..fc9ab8ec2 100644 --- a/java/API.md +++ b/java/API.md @@ -15,7 +15,7 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G | `JobUpdate` output parameter, `RecordOutput` | `Client.output`, `WorkContext.output` | Names describe the supported operation; attempt output is buffered until completion. | | `JobCompleteTx` | `Client.complete(connection, job)`, `WorkContext.complete(connection)` | Completion keeps `Job` and commits with application changes. Only a running attempt can be completed. A committed completion remains authoritative if the handler subsequently throws. | | Context client, cancellation, cancel/snooze errors | `WorkContext.client`, cancellation methods, `cancel`, `discard`, `snooze` | Java has no Go context parameter convention. Control methods end the attempt; virtual-thread interruption supplements cooperative cancellation. | -| Hooks and middleware | `Extension` | Default methods group related callbacks. JDBC hooks and middleware may throw checked exceptions; insert middleware receives a standard `Callable`. | +| Plugins, hooks, and middleware | `Plugin`, `HookInsertBegin`, `HookPeriodicJobsStart`, `HookWorkBegin`, `HookWorkEnd`, `JobInsertMiddleware`, `WorkerMiddleware` | `withPlugins` activates every hook and middleware interface on a value. `withHooks` and `withMiddleware` restrict registration to that side. Plugins run first, as in Go. Functional interfaces support lambdas and checked exceptions. | | Retry policy | `RetryPolicy` | A lambda receives the full job and returns `Duration`; the default uses the same error-count backoff and jitter. | | `Queues`, queue CRUD/control APIs | `Workers.addQueue/removeQueue`, `Client.queues()` | Runtime concurrency and persisted queue state have separate owners. Reads and mutations support caller transactions. | | `Subscribe`, `EventKind`, `Event` | `Workers.subscribe`, `EventKind`, `Event`, `Subscription` | Enum filters and an `AutoCloseable` subscription replace channels. Callbacks are synchronous; queue events include their queue. | @@ -27,6 +27,14 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G ## Changes made before release +- Replaced `Extension` and `withExtension` with separate hook and middleware + interfaces. Names follow Go, while functional interfaces allow Java lambdas. + Work-end hooks can preserve, replace, or clear the failure; a work-begin failure + skips the handler and work-end hooks. The conformance claim barrier now uses + worker middleware instead of an application-facing `afterClaim` callback. +- Added the common `Plugin` marker and `withPlugins` registration to install all + of an implementation's hooks and middleware together. Renamed the internal Pro + storage interface and registration from `Plugin`/`withPlugin` to `Pilot`/`withPilot`. - Removed no-op `Client.close()`. Close `Workers` and the application-owned pool. - Added typed retrieval and typed transactional completion, including running-state validation. - Unified bulk insertion around typed homogeneous lists and mixed-kind submissions, @@ -44,7 +52,7 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G - Changed retry callbacks to receive the full job, enabling kind- and metadata-specific policies. - Added named migration options and fluent uniqueness state/kind overrides. - Kept destructive reset and retention helpers behind the internal driver seam. - `Plugin`, `Client.Driver`, protocol JSON helpers, and explicitly marked companion + `Pilot`, `Client.Driver`, protocol JSON helpers, and explicitly marked companion methods are not stable application extension APIs. Options copy collections and metadata on construction; `InsertOptions.metadata()` diff --git a/java/README.md b/java/README.md index b6b14557c..8e6fe5874 100644 --- a/java/README.md +++ b/java/README.md @@ -285,7 +285,7 @@ requested. Call `context.checkCancelled()` when abandoning unfinished work. ### Getting the client within workers [`context.client()`](https://riverqueue.com/docs/context-client) gives a handler -its client, including configured extensions. Use its transaction overloads when +its client, including configured hooks and middleware. Use its transaction overloads when enqueueing follow-up work that must commit with other writes or job completion. ```java @@ -300,22 +300,21 @@ workerConfig.add(welcome, context -> Java exceptions and uncaught `Error`s are handled as failed attempts; River persists failures and retries according to policy. For [application error reporting](https://riverqueue.com/docs/error-handling), attach -an `Extension.afterWork` hook. `Workers.Builder.errorHandler` separately receives +a `HookWorkEnd` hook. `Workers.Builder.errorHandler` separately receives runtime failures, such as database or subscriber errors. Failures in observers or event subscribers are reported without changing the job's outcome or preventing other callbacks from running. If the runtime error handler itself throws, River logs both failures and continues. ```java -var monitoredClient = client.withExtension(new Extension() { - @Override - public void afterWork(WorkContext context, Throwable failure) { - if (failure != null) { - System.getLogger("jobs").log(System.Logger.Level.ERROR, - "Job " + context.job().id() + " failed", failure); - } +HookWorkEnd reportFailure = (context, failure) -> { + if (failure != null) { + System.getLogger("jobs").log(System.Logger.Level.ERROR, + "Job " + context.job().id() + " failed", failure); } -}); + return failure; +}; +var monitoredClient = client.withPlugins(reportFailure); var workerConfig = monitoredClient.workers() .queue("default", 10) .errorHandler(failure -> @@ -326,6 +325,88 @@ var workerConfig = monitoredClient.workers() Hooks also see the control exceptions used for snoozing and cancellation. Keep reporting hooks fast and avoid throwing from them. +### Plugins, hooks, and middleware + +Use `client.withPlugins(...)` to install hooks and middleware together. `Plugin` +is their common marker interface: `Hook` and `Middleware` both extend it. A +plugin automatically activates every operation-specific interface it implements. + +Hooks observe individual lifecycle stages through `HookInsertBegin`, +`HookWorkBegin`, `HookWorkEnd`, and `HookPeriodicJobsStart`. Middleware wraps an +operation with a `next` callback through `JobInsertMiddleware` and +`WorkerMiddleware`. Each is a functional interface for lambdas or method +references. For example, individual callbacks can be installed together: + +```java +JobInsertMiddleware logInsert = (connection, submissions, next) -> { + var results = next.call(); + System.getLogger("jobs").log(System.Logger.Level.DEBUG, + "Inserted " + results.size() + " jobs"); + return results; +}; +WorkerMiddleware timeWork = (context, next) -> { + long started = System.nanoTime(); + try { + next.run(); + } finally { + System.getLogger("jobs").log(System.Logger.Level.DEBUG, + "Job " + context.job().id() + " took " + (System.nanoTime() - started) + " ns"); + } +}; +var instrumentedClient = client.withPlugins(reportFailure, logInsert, timeWork); +``` + +A reusable plugin can implement multiple interfaces on one class: + +```java +final class JobInstrumentation implements HookWorkEnd, WorkerMiddleware { + @Override + public void work(WorkContext context, WorkContext.Step next) throws Exception { + System.getLogger("jobs").log(System.Logger.Level.DEBUG, + "Starting job " + context.job().id()); + next.run(); + } + + @Override + public Throwable workEnd(WorkContext context, Throwable failure) { + if (failure != null) { + System.getLogger("jobs").log(System.Logger.Level.ERROR, + "Job " + context.job().id() + " failed", failure); + } + return failure; + } +} + +var instrumentedClient = client.withPlugins(new JobInstrumentation()); +``` + +Use `withHooks(...)` or `withMiddleware(...)` when only that side of an +implementation should be activated. All three registration methods return a new +client and append to their respective lists. Use the returned client for insertion +and worker configuration. Instances can run concurrently and must protect shared +mutable state. + +As in Go, plugins run first, followed by separately registered hooks or middleware, +regardless of which registration method was called first. Each list preserves its +registration order. Repeated registrations are retained, so registering the same +instance twice invokes it twice for each applicable phase. + +Hooks run in registration order inside the innermost middleware. Middleware runs +in registration order before `next` and unwinds in reverse order afterward. +Insertion middleware runs once per batch, including a one-element batch for a +single insertion; insert hooks run once per job. Middleware receives an immutable +list of submissions with the caller's arguments and option overrides. Insert +hooks and middleware run in the insertion's transaction, and a failure anywhere +rolls back the insertion, even after `next` returns. + +Work hooks and middleware receive decoded arguments. A work-begin failure skips +the handler and work-end hooks. Work-end hooks receive the handler's exception +or `Error`, or `null` for success, and return the outcome passed to the next hook. +Return the received failure to preserve it, a different failure to replace it, +or `null` to report success. A thrown exception skips remaining work-end hooks. +The internal `Pilot` interface is a River Pro storage integration point, separate +from application plugins. + ### Job-persisted logging [Job logs](https://riverqueue.com/docs/job-logging) live in `river:log` metadata diff --git a/java/conformance/feature-inventory.json b/java/conformance/feature-inventory.json index fab7a70a8..09d0c276c 100644 --- a/java/conformance/feature-inventory.json +++ b/java/conformance/feature-inventory.json @@ -504,7 +504,7 @@ "id": "config.AdvisoryLockPrefix", "rationale": "Copied into the periodic job enqueuer's configuration but not used to derive any lock key in this version; it has no persisted or cross-process effect to match.", "source": "client.go:river.Config.AdvisoryLockPrefix", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -516,7 +516,7 @@ "maintenance_job_cleaner_retention" ], "source": "client.go:river.Config.CancelledJobRetentionPeriod", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -528,7 +528,7 @@ "maintenance_job_cleaner_retention" ], "source": "client.go:river.Config.CompletedJobRetentionPeriod", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -540,7 +540,7 @@ "maintenance_job_cleaner_retention" ], "source": "client.go:river.Config.DiscardedJobRetentionPeriod", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -552,7 +552,7 @@ "error_handler_cancel_override" ], "source": "client.go:river.Config.ErrorHandler", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -561,7 +561,7 @@ "id": "config.FetchCooldown", "rationale": "Per-client minimum interval between fetches (a throughput throttle), which also suppresses a client's repeated insert notification for a queue within the interval on every backend. Implementations expose an equivalent client-level knob with the same default and minimum. Rows are unaffected; the reference adapter's 1 ms setting keeps notification scenarios deterministic.", "source": "client.go:river.Config.FetchCooldown", - "java_surface": "Workers.Builder, JobType, Extension, Database", + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database", "java_difference": "Wakeups are coalesced; Java does not expose a separate fetch cooldown knob." }, { @@ -571,7 +571,7 @@ "id": "config.FetchOnlyKnownKinds", "rationale": "Restricts a client's claims to the kinds of its registered workers, including aliases, so jobs of other kinds stay available without using attempts. The filter is local to the claiming client; implementations expose an equivalent option, and the jobs it leaves are ordinary available rows any client may work.", "source": "client.go:river.Config.FetchOnlyKnownKinds", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -584,7 +584,7 @@ "notification_only_wakeups" ], "source": "client.go:river.Config.FetchPollInterval", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -593,7 +593,7 @@ "id": "config.Hooks", "rationale": "Registration of global hooks in each language's idiom. Hook ordering semantics are exercised through plugin registration in extension_hook_middleware_order.", "source": "client.go:river.Config.Hooks", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Client.withHooks" }, { "applicability": "protocol_visible", @@ -606,7 +606,7 @@ "sqlite_runtime_attempted_by_ordering" ], "source": "client.go:river.Config.ID", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "internal", @@ -615,7 +615,7 @@ "id": "config.JobCleanerTimeout", "rationale": "Timeout for individual job cleaner queries; bounds local work only and changes no persisted outcome.", "source": "client.go:river.Config.JobCleanerTimeout", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "not_applicable", @@ -624,7 +624,7 @@ "id": "config.JobInsertMiddleware", "rationale": "Deprecated Go field superseded by Plugins. The insert-middleware concept is classified under extension.rivertype.JobInsertMiddleware.", "source": "client.go:river.Config.JobInsertMiddleware", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -636,7 +636,7 @@ "stuck_job_detection" ], "source": "client.go:river.Config.JobStuckHandler", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -648,7 +648,7 @@ "stuck_job_detection" ], "source": "client.go:river.Config.JobStuckThreshold", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -660,7 +660,7 @@ "timeout_cancellation" ], "source": "client.go:river.Config.JobTimeout", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -674,7 +674,7 @@ "sqlite_runtime_leader_election_disabled" ], "source": "client.go:river.Config.LeaderElectionDisabled", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -683,7 +683,7 @@ "id": "config.Logger", "rationale": "Each implementation uses its own logging facility.", "source": "client.go:river.Config.Logger", - "java_surface": "Workers.Builder, JobType, Extension, Database", + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database", "java_difference": "Uses System.Logger and Workers.Builder.errorHandler instead of a logger object on the client." }, { @@ -697,7 +697,7 @@ "reference_insert_candidate_work" ], "source": "client.go:river.Config.MaxAttempts", - "java_surface": "Workers.Builder, JobType, Extension, Database", + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database", "java_difference": "Configure insert defaults on JobType instead of a process-wide client default." }, { @@ -707,7 +707,7 @@ "id": "config.Middleware", "rationale": "Registration of global middleware in each language's idiom. Middleware ordering semantics are exercised through plugin registration in extension_hook_middleware_order.", "source": "client.go:river.Config.Middleware", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Client.withMiddleware" }, { "applicability": "protocol_visible", @@ -722,7 +722,7 @@ "sqlite_runtime_periodic_scheduler" ], "source": "client.go:river.Config.PeriodicJobs", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -734,7 +734,7 @@ "extension_hook_middleware_order" ], "source": "client.go:river.Config.Plugins", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Client.withPlugins" }, { "applicability": "api_equivalent", @@ -748,7 +748,7 @@ "sqlite_runtime_poll_only_recovery" ], "source": "client.go:river.Config.PollOnly", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -761,7 +761,7 @@ "dynamic_queue_add_reconfigure_remove" ], "source": "client.go:river.Config.Queues", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -773,7 +773,7 @@ "maintenance_reindexer_skips_artifacts" ], "source": "client.go:river.Config.ReindexerIndexNames", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -785,7 +785,7 @@ "maintenance_reindexer_skips_artifacts" ], "source": "client.go:river.Config.ReindexerSchedule", - "java_surface": "Workers.Builder, JobType, Extension, Database", + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database", "java_difference": "Supports a duration override or the default daily UTC schedule; arbitrary reindex cron schedules are not exposed." }, { @@ -795,7 +795,7 @@ "id": "config.ReindexerTimeout", "rationale": "Per-reindex operation timeout; bounds local work only.", "source": "client.go:river.Config.ReindexerTimeout", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -809,7 +809,7 @@ "reference_process_kill_candidate_rescue" ], "source": "client.go:river.Config.RescueStuckJobsAfter", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -822,7 +822,7 @@ "deterministic_retry_clock_rng" ], "source": "client.go:river.Config.RetryPolicy", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "protocol_visible", @@ -835,7 +835,7 @@ "custom_schema_reference_migrate_candidate_work" ], "source": "client.go:river.Config.Schema", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -844,7 +844,7 @@ "id": "config.SkipJobKindValidation", "rationale": "Deprecated escape hatch that skips kind-format validation at insert time; implementations may offer an equivalent legacy-kind option.", "source": "client.go:river.Config.SkipJobKindValidation", - "java_surface": "Workers.Builder, JobType, Extension, Database", + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database", "java_difference": "New JobType values require the Go-compatible kind format; no legacy validation bypass is exposed." }, { @@ -857,7 +857,7 @@ "mixed_unknown_kind_error" ], "source": "client.go:river.Config.SkipUnknownJobCheck", - "java_surface": "Workers.Builder, JobType, Extension, Database", + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database", "java_difference": "Inserts are independent of worker registrations, which permits inserting jobs owned by other languages." }, { @@ -870,7 +870,7 @@ "hard_shutdown_soft_stop_classification" ], "source": "client.go:river.Config.SoftStopTimeout", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "not_applicable", @@ -879,7 +879,7 @@ "id": "config.Test", "rationale": "Go test-environment settings (time generator, unique enforcement toggle). Conformance drives time through the adapter's clock_set instead.", "source": "client.go:river.Config.Test", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "not_applicable", @@ -888,7 +888,7 @@ "id": "config.TestOnly", "rationale": "Go test-suite switch that removes startup jitter; not part of any production behavior.", "source": "client.go:river.Config.TestOnly", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "not_applicable", @@ -897,7 +897,7 @@ "id": "config.WorkerMiddleware", "rationale": "Deprecated Go field superseded by Plugins. The worker-middleware concept is classified under extension.rivertype.WorkerMiddleware.", "source": "client.go:river.Config.WorkerMiddleware", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "api_equivalent", @@ -906,7 +906,7 @@ "id": "config.Workers", "rationale": "Language-native worker registry mapping kinds to handlers.", "source": "client.go:river.Config.Workers", - "java_surface": "Workers.Builder, JobType, Extension, Database" + "java_surface": "Workers.Builder, JobType, Hook, Middleware, Database" }, { "applicability": "driver_specific", @@ -1851,7 +1851,7 @@ "id": "extension.riverpilot.Pilot.JobCancel", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.JobCancel", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1860,7 +1860,7 @@ "id": "extension.riverpilot.Pilot.JobCleanerQueuesExcluded", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.JobCleanerQueuesExcluded", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1869,7 +1869,7 @@ "id": "extension.riverpilot.Pilot.JobGetAvailable", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.JobGetAvailable", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1878,7 +1878,7 @@ "id": "extension.riverpilot.Pilot.JobInsertMany", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.JobInsertMany", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1887,7 +1887,7 @@ "id": "extension.riverpilot.Pilot.JobRetry", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.JobRetry", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1896,7 +1896,7 @@ "id": "extension.riverpilot.Pilot.JobSetStateIfRunningMany", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.JobSetStateIfRunningMany", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1905,7 +1905,7 @@ "id": "extension.riverpilot.Pilot.PilotInit", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.PilotInit", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1914,7 +1914,7 @@ "id": "extension.riverpilot.Pilot.PilotPeriodicJob", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1923,7 +1923,7 @@ "id": "extension.riverpilot.Pilot.ProducerInit", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.ProducerInit", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1932,7 +1932,7 @@ "id": "extension.riverpilot.Pilot.ProducerKeepAlive", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.ProducerKeepAlive", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1941,7 +1941,7 @@ "id": "extension.riverpilot.Pilot.ProducerShutdown", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.ProducerShutdown", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1950,7 +1950,7 @@ "id": "extension.riverpilot.Pilot.QueueMetadataChanged", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.Pilot.QueueMetadataChanged", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1959,7 +1959,7 @@ "id": "extension.riverpilot.PilotJobCompletionConcurrency.JobSetStateIfRunningManyConcurrency", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.PilotJobCompletionConcurrency.JobSetStateIfRunningManyConcurrency", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1968,7 +1968,7 @@ "id": "extension.riverpilot.PilotJobRescuer.JobGetStuck", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.PilotJobRescuer.JobGetStuck", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1977,7 +1977,7 @@ "id": "extension.riverpilot.PilotJobRescuer.JobRescueMany", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.PilotJobRescuer.JobRescueMany", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1986,7 +1986,7 @@ "id": "extension.riverpilot.PilotPeriodicJob.PeriodicJobGetAll", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.PilotPeriodicJob.PeriodicJobGetAll", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -1995,7 +1995,7 @@ "id": "extension.riverpilot.PilotPeriodicJob.PeriodicJobKeepAliveAndReap", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.PilotPeriodicJob.PeriodicJobKeepAliveAndReap", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -2004,7 +2004,7 @@ "id": "extension.riverpilot.PilotPeriodicJob.PeriodicJobUpsertMany", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.PilotPeriodicJob.PeriodicJobUpsertMany", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "internal", @@ -2013,7 +2013,7 @@ "id": "extension.riverpilot.ProducerState.JobFinish", "rationale": "Part of the unstable Go extension seam (riverpilot) used to substitute storage operations. Implementations may have their own seam; it carries no cross-language contract, and the default behavior's effects are covered by the driver and protocol items.", "source": "rivershared/riverpilot/pilot.go:riverpilot.ProducerState.JobFinish", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Pilot and Client.Driver (internal Pro seam)" }, { "applicability": "not_applicable", @@ -2022,7 +2022,7 @@ "id": "extension.rivertype.Hook.IsHook", "rationale": "Go marker method used to discriminate hook, middleware, and plugin values; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.Hook.IsHook", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Hook" }, { "applicability": "not_applicable", @@ -2031,7 +2031,7 @@ "id": "extension.rivertype.HookInsertBegin.Hook", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.HookInsertBegin", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookInsertBegin" }, { "applicability": "api_equivalent", @@ -2043,7 +2043,7 @@ "extension_hook_middleware_order" ], "source": "rivertype/river_type.go:rivertype.HookInsertBegin.InsertBegin", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookInsertBegin" }, { "applicability": "not_applicable", @@ -2052,7 +2052,7 @@ "id": "extension.rivertype.HookMetricEmit.Hook", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.HookMetricEmit", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Plugin, Hook, Middleware, and Pilot (internal Pro seam)" }, { "applicability": "not_applicable", @@ -2061,7 +2061,7 @@ "id": "extension.rivertype.HookMetricEmit.MetricEmit", "rationale": "Go-specific metric hook; other implementations expose telemetry through their own instrumentation.", "source": "rivertype/river_type.go:rivertype.HookMetricEmit.MetricEmit", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Plugin, Hook, Middleware, and Pilot (internal Pro seam)" }, { "applicability": "not_applicable", @@ -2070,7 +2070,7 @@ "id": "extension.rivertype.HookPeriodicJobsStart.Hook", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.HookPeriodicJobsStart", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookPeriodicJobsStart" }, { "applicability": "api_equivalent", @@ -2082,7 +2082,7 @@ "periodic_run_on_start" ], "source": "rivertype/river_type.go:rivertype.HookPeriodicJobsStart.Start", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookPeriodicJobsStart" }, { "applicability": "not_applicable", @@ -2091,7 +2091,7 @@ "id": "extension.rivertype.HookWorkBegin.Hook", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.HookWorkBegin", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookWorkBegin" }, { "applicability": "api_equivalent", @@ -2103,7 +2103,7 @@ "extension_hook_middleware_order" ], "source": "rivertype/river_type.go:rivertype.HookWorkBegin.WorkBegin", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookWorkBegin" }, { "applicability": "not_applicable", @@ -2112,7 +2112,7 @@ "id": "extension.rivertype.HookWorkEnd.Hook", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.HookWorkEnd", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookWorkEnd" }, { "applicability": "api_equivalent", @@ -2124,7 +2124,7 @@ "extension_hook_middleware_order" ], "source": "rivertype/river_type.go:rivertype.HookWorkEnd.WorkEnd", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "HookWorkEnd" }, { "applicability": "api_equivalent", @@ -2136,7 +2136,7 @@ "extension_hook_middleware_order" ], "source": "rivertype/river_type.go:rivertype.JobInsertMiddleware.InsertMany", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "JobInsertMiddleware" }, { "applicability": "not_applicable", @@ -2145,7 +2145,7 @@ "id": "extension.rivertype.JobInsertMiddleware.Middleware", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.JobInsertMiddleware", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "JobInsertMiddleware" }, { "applicability": "not_applicable", @@ -2154,7 +2154,7 @@ "id": "extension.rivertype.Middleware.IsMiddleware", "rationale": "Go marker method used to discriminate hook, middleware, and plugin values; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.Middleware.IsMiddleware", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Middleware" }, { "applicability": "not_applicable", @@ -2163,7 +2163,7 @@ "id": "extension.rivertype.Plugin.IsPlugin", "rationale": "Go marker method used to discriminate hook, middleware, and plugin values; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.Plugin.IsPlugin", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "Plugin (marker interface inherited by Hook and Middleware)" }, { "applicability": "not_applicable", @@ -2172,7 +2172,7 @@ "id": "extension.rivertype.WorkerMiddleware.Middleware", "rationale": "Go interface composition marking the value as a hook or middleware; other languages express this with their own type systems.", "source": "rivertype/river_type.go:rivertype.WorkerMiddleware", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "WorkerMiddleware" }, { "applicability": "api_equivalent", @@ -2184,7 +2184,7 @@ "extension_hook_middleware_order" ], "source": "rivertype/river_type.go:rivertype.WorkerMiddleware.Work", - "java_surface": "Extension and Plugin (internal Pro seam)" + "java_surface": "WorkerMiddleware" }, { "applicability": "protocol_visible", diff --git a/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java b/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java index e91a1c4e9..5a70cd7e8 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java @@ -71,7 +71,7 @@ private Adapter() throws Exception { applicationName); pool = Connections.pool(source); database = new Database(pool, source.dialect()); - river = new Client(database).withExtension(runtime.instrumentation()); + river = runtime.instrument(new Client(database)); try (var input = Adapter.class.getResourceAsStream("/contract.json")) { contract = Json.MAPPER.readTree(input); } @@ -266,7 +266,7 @@ private JsonNode handle(String method, JsonNode params, String rawParams) throws case "reset" -> { runtime.stop(true); river.driver().reset(); - river = new Client(database).withExtension(runtime.instrumentation()); + river = runtime.instrument(new Client(database)); yield Json.object(); } case "get" -> normalized(scoped(params).get(params.path("id").asLong())); @@ -509,8 +509,7 @@ static JsonNode normalized(Job job) { private Client scoped(JsonNode params) { return params.path("schema").asString("").isEmpty() ? river - : new Client(database.withSchema(params.path("schema").asString())) - .withExtension(runtime.instrumentation()); + : runtime.instrument(new Client(database.withSchema(params.path("schema").asString()))); } private static boolean fitsDouble(JsonNode value) { 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 1607678ec..be404345a 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java @@ -1,15 +1,20 @@ package com.riverqueue.conformance; import com.riverqueue.Client; -import com.riverqueue.Extension; +import com.riverqueue.HookInsertBegin; +import com.riverqueue.HookPeriodicJobsStart; +import com.riverqueue.HookWorkBegin; +import com.riverqueue.HookWorkEnd; import com.riverqueue.InsertOptions; import com.riverqueue.Job; +import com.riverqueue.JobInsertMiddleware; import com.riverqueue.JobType; import com.riverqueue.Json; import com.riverqueue.RiverException; import com.riverqueue.Schedule; import com.riverqueue.Unique; import com.riverqueue.WorkContext; +import com.riverqueue.WorkerMiddleware; import com.riverqueue.Workers; import java.sql.Connection; import java.time.Duration; @@ -82,53 +87,62 @@ JsonNode handle(Client river, String method, JsonNode params) throws Exception { }; } - Extension instrumentation() { - return new Extension() { - @Override - public void afterClaim(Job job) throws Exception { - if (claimBarrier != null && firstClaim.compareAndSet(false, true)) claimBarrier.await(); - } + Client instrument(Client river) { + return river.withPlugins(new Instrumentation()); + } - @Override - public void beforeInsert( - Connection connection, JobType type, Object args, InsertOptions options) { - if (probe.instrumented) probe.trace.add("hook:insert_begin"); - } + private final class Instrumentation + implements HookInsertBegin, + HookPeriodicJobsStart, + HookWorkBegin, + HookWorkEnd, + JobInsertMiddleware, + WorkerMiddleware { + @Override + public void insertBegin( + Connection connection, JobType type, Object args, InsertOptions options) { + if (probe.instrumented) probe.trace.add("hook:insert_begin"); + } - @Override - public void beforeWork(WorkContext context) { - if (probe.instrumented) probe.trace.add("hook:work_begin"); - } + @Override + public List insertMany( + Connection connection, + List> submissions, + Callable> next) + throws Exception { + if (probe.instrumented) probe.trace.add("middleware:insert_before"); + var result = next.call(); + if (probe.instrumented) probe.trace.add("middleware:insert_after"); + return result; + } - @Override - public void afterWork(WorkContext context, Throwable failure) { - if (probe.instrumented) probe.trace.add("hook:work_end"); - } + @Override + public void start() { + probe.periodicStarts.incrementAndGet(); + if (probe.instrumented) probe.trace.add("hook:periodic_start"); + } - @Override - public T insert(Connection connection, Callable next) throws Exception { - if (probe.instrumented) probe.trace.add("middleware:insert_before"); - T result = next.call(); - if (probe.instrumented) probe.trace.add("middleware:insert_after"); - return result; + @Override + public void work(WorkContext context, WorkContext.Step next) throws Exception { + if (claimBarrier != null && firstClaim.compareAndSet(false, true)) claimBarrier.await(); + if (probe.instrumented) probe.trace.add("middleware:work_before"); + try { + next.run(); + } finally { + if (probe.instrumented) probe.trace.add("middleware:work_after"); } + } - @Override - public void periodicStarted() { - probe.periodicStarts.incrementAndGet(); - if (probe.instrumented) probe.trace.add("hook:periodic_start"); - } + @Override + public void workBegin(WorkContext context) { + if (probe.instrumented) probe.trace.add("hook:work_begin"); + } - @Override - public void work(WorkContext context, WorkContext.Step next) throws Exception { - if (probe.instrumented) probe.trace.add("middleware:work_before"); - try { - next.run(); - } finally { - if (probe.instrumented) probe.trace.add("middleware:work_after"); - } - } - }; + @Override + public Throwable workEnd(WorkContext context, Throwable failure) { + if (probe.instrumented) probe.trace.add("hook:work_end"); + return failure; + } } void stop(boolean cancel) { diff --git a/java/river/src/main/java/com/riverqueue/Client.java b/java/river/src/main/java/com/riverqueue/Client.java index a77fb5c7b..19db1098d 100644 --- a/java/river/src/main/java/com/riverqueue/Client.java +++ b/java/river/src/main/java/com/riverqueue/Client.java @@ -10,6 +10,7 @@ import java.util.HexFormat; import java.util.List; import java.util.Objects; +import java.util.stream.Stream; import tools.jackson.databind.JsonNode; /** @@ -21,24 +22,53 @@ public final class Client { private final java.util.concurrent.CopyOnWriteArrayList insertListeners = new java.util.concurrent.CopyOnWriteArrayList<>(); private final ThreadLocal ownedCommit = new ThreadLocal<>(); - private final Plugin plugin; + private final Pilot pilot; private final Clock clock; private final Database database; - private final List extensions; + private final List allHooks; + private final List hooks; + private final List insertMiddleware; + private final List middleware; + private final List plugins; + private final List workerMiddleware; public Client(Database database) { this(database, Clock.systemUTC()); } public Client(Database database, Clock clock) { - this(database, clock, List.of(), new Plugin() {}); + this(database, clock, List.of(), List.of(), List.of(), new Pilot() {}); } - private Client(Database database, Clock clock, List extensions, Plugin plugin) { - this.plugin = plugin; + private Client( + Database database, + Clock clock, + List hooks, + List middleware, + List plugins, + Pilot pilot) { + this.pilot = pilot; this.database = Objects.requireNonNull(database); this.clock = Objects.requireNonNull(clock); - this.extensions = List.copyOf(extensions); + this.hooks = List.copyOf(hooks); + this.middleware = List.copyOf(middleware); + this.plugins = List.copyOf(plugins); + this.allHooks = + Stream.concat(this.plugins.stream(), this.hooks.stream()) + .filter(Hook.class::isInstance) + .map(Hook.class::cast) + .toList(); + var allMiddleware = Stream.concat(this.plugins.stream(), this.middleware.stream()).toList(); + this.insertMiddleware = + allMiddleware.stream() + .filter(JobInsertMiddleware.class::isInstance) + .map(JobInsertMiddleware.class::cast) + .toList(); + this.workerMiddleware = + allMiddleware.stream() + .filter(WorkerMiddleware.class::isInstance) + .map(WorkerMiddleware.class::cast) + .toList(); } public Database database() { @@ -53,15 +83,42 @@ public Workers.Builder workers() { return new Workers.Builder(this); } - public Client withExtension(Extension extension) { - var all = new ArrayList<>(extensions); - all.add(extension); - return new Client(database, clock, all, plugin); + /** + * Returns a new client with hooks appended in registration order, after plugins. Only hook + * interfaces are activated; use {@link #withPlugins(Plugin...)} to also activate middleware. + */ + public Client withHooks(Hook... hooks) { + var all = new ArrayList<>(this.hooks); + all.addAll(List.of(hooks)); + return new Client(database, clock, all, middleware, plugins, pilot); + } + + /** + * Returns a new client with middleware appended after plugins; the first middleware is outermost. + * Only middleware interfaces are activated; use {@link #withPlugins(Plugin...)} to also activate + * hooks. + */ + public Client withMiddleware(Middleware... middleware) { + var all = new ArrayList<>(this.middleware); + all.addAll(List.of(middleware)); + return new Client(database, clock, hooks, all, plugins, pilot); } /** Internal integration point used by River Pro; not a stable application API. */ - public Client withPlugin(Plugin plugin) { - return new Client(database, clock, extensions, Objects.requireNonNull(plugin)); + public Client withPilot(Pilot pilot) { + return new Client(database, clock, hooks, middleware, plugins, Objects.requireNonNull(pilot)); + } + + /** + * Returns a new client with plugins appended in registration order. Each plugin activates all its + * hook and middleware interfaces. Plugins run before separately configured hooks and middleware, + * regardless of which registration method was called first. Repeated registrations are retained, + * including when the same instance is registered through multiple methods. + */ + public Client withPlugins(Plugin... plugins) { + var all = new ArrayList<>(this.plugins); + all.addAll(List.of(plugins)); + return new Client(database, clock, hooks, middleware, all, pilot); } Runnable onInsertCommit(Runnable listener) { @@ -78,8 +135,8 @@ private static final class Commit { } } - Plugin plugin() { - return plugin; + Pilot pilot() { + return pilot; } /** Internal driver access for the matched River Pro release. */ @@ -157,18 +214,24 @@ public Job.InsertResult reinsert(Connection connection, Job original) connection, c -> insertMiddleware( - c, - 0, - () -> { - for (var extension : extensions) - extension.beforeInsert(c, type, original.args(), options); - return insertOne(c, type, original.args(), options, original, null); - })); + c, + List.of(type.submission(original.args(), options)), + 0, + () -> { + insertBegin(c, type, original.args(), options); + return List.of( + insertOne(c, type, original.args(), options, original, null)); + }) + .getFirst()); } } - List extensions() { - return extensions; + List hooks() { + return allHooks; + } + + List workerMiddleware() { + return workerMiddleware; } public Job cancel(long id) { @@ -191,7 +254,7 @@ public Job cancel(Connection connection, long id) { var rows = statement.executeQuery()) { if (!rows.next()) return get(c, id); var job = read(rows); - plugin.afterStateChange(c, driver(), job); + pilot.afterStateChange(c, driver(), job); notify(c, Protocol.cancel(id, job.queue())); return job; } @@ -213,7 +276,7 @@ public Job delete(Connection connection, long id) { var rows = statement.executeQuery()) { rows.next(); var deleted = read(rows); - plugin.afterDelete(c, driver(), deleted); + pilot.afterDelete(c, driver(), deleted); return deleted; } }); @@ -274,7 +337,7 @@ public Job complete(Connection connection, Job job, Object metadata) { var rows = statement.executeQuery()) { if (!rows.next()) return get(c, job.id()).mapArgs(ignored -> job.args()); var completed = read(rows); - plugin.afterStateChange(c, driver(), completed); + pilot.afterStateChange(c, driver(), completed); return completed.mapArgs(ignored -> job.args()); } }); @@ -313,7 +376,7 @@ public List> deleteMany(Connection connection, JobQuery query, boo // Recheck every filter atomically: a peer may have updated this job since listing. if (!rows.next()) continue; var deleted = read(rows); - plugin.afterDelete(c, driver(), deleted); + pilot.afterDelete(c, driver(), deleted); result.add(deleted); } } @@ -382,22 +445,38 @@ Job.InsertResult insert( connection, c -> { return insertMiddleware( - c, - 0, - () -> { - for (var extension : extensions) extension.beforeInsert(c, type, args, options); - return insertOne(c, type, args, options, null, scheduledDefault); - }); + c, + List.of(type.submission(args, options)), + 0, + () -> { + insertBegin(c, type, args, options); + return List.of(insertOne(c, type, args, options, null, scheduledDefault)); + }) + .getFirst(); }); } - private T insertMiddleware( - Connection connection, int index, java.util.concurrent.Callable next) throws Exception { - return index == extensions.size() + private void insertBegin( + Connection connection, JobType type, Object args, InsertOptions options) throws Exception { + for (var hook : allHooks) + if (hook instanceof HookInsertBegin insertHook) + insertHook.insertBegin(connection, type, args, options); + } + + private List insertMiddleware( + Connection connection, + List> submissions, + int index, + java.util.concurrent.Callable> next) + throws Exception { + return index == insertMiddleware.size() ? next.call() - : extensions + : insertMiddleware .get(index) - .insert(connection, () -> insertMiddleware(connection, index + 1, next)); + .insertMany( + connection, + submissions, + () -> insertMiddleware(connection, submissions, index + 1, next)); } /** Inserts jobs of one kind atomically, inheriting the job type's defaults. */ @@ -420,7 +499,7 @@ public List insertMany( /** Inserts jobs of one kind with common options in the caller's transaction. */ public List insertMany( Connection connection, JobType type, List args, InsertOptions options) { - return insertBatch(connection, args, (c, arg) -> insertWithHooks(c, type, arg, options)); + return insertMany(connection, args.stream().map(arg -> type.submission(arg, options)).toList()); } /** Inserts mixed job kinds and per-job options atomically, in input order. */ @@ -431,13 +510,13 @@ public List insertMany(List> submissio /** Inserts mixed job kinds and per-job options inside the caller's transaction. */ public List insertMany( Connection connection, List> submissions) { - return insertBatch(connection, submissions, (c, submission) -> insertSubmission(c, submission)); + return insertBatch(connection, List.copyOf(submissions)); } private Job.InsertResult insertWithHooks( Connection connection, JobType type, A args, InsertOptions options) { try { - for (var extension : extensions) extension.beforeInsert(connection, type, args, options); + insertBegin(connection, type, args, options); } catch (Exception error) { throw propagate(error); } @@ -448,22 +527,21 @@ private Job.InsertResult insertSubmission(Connection connection, Submission< return insertWithHooks(connection, value.type(), value.args(), value.options()); } - private List insertBatch( - Connection connection, - List values, - java.util.function.BiFunction insert) { - if (values.isEmpty()) throw new IllegalArgumentException("Cannot insert an empty batch"); + private List insertBatch( + Connection connection, List> submissions) { + if (submissions.isEmpty()) throw new IllegalArgumentException("Cannot insert an empty batch"); return atomic( connection, c -> insertMiddleware( c, + submissions, 0, () -> { var results = new ArrayList(); var keys = new java.util.HashSet(); - for (var value : values) { - var result = insert.apply(c, value); + for (var submission : submissions) { + var result = insertSubmission(c, submission); if (result.job().uniqueKey() != null && !keys.add(result.job().uniqueKey())) throw new RiverException( RiverException.Code.REJECTED, "Batch repeats a unique key"); @@ -490,7 +568,7 @@ private Job.InsertResult insertOne( String encodedArgs = Json.encode(args); try { var prepared = - plugin.prepare(connection, driver(), new Plugin.Insert(type, encodedArgs, resolved)); + pilot.prepare(connection, driver(), new Pilot.Insert(type, encodedArgs, resolved)); encodedArgs = prepared.args(); resolved = prepared.options(); validate(resolved); @@ -585,7 +663,7 @@ private Job.InsertResult insertOne( .path(Protocol.METADATA_UNIQUE_NONCE) .equals(metadata.path(Protocol.METADATA_UNIQUE_NONCE)) : rows.getBoolean("duplicate"); - if (!duplicate) plugin.afterInsert(connection, driver(), job); + if (!duplicate) pilot.afterInsert(connection, driver(), job); if (!duplicate && job.state() == Job.State.AVAILABLE) notifyInsert(connection, job.queue()); return new Job.InsertResult(job, duplicate); } finally { @@ -709,7 +787,7 @@ public Job retry(Connection connection, long id) { var rows = statement.executeQuery()) { boolean updated = rows.next(); var job = updated ? read(rows) : before; - plugin.afterStateChange(c, driver(), job); + pilot.afterStateChange(c, driver(), job); if (updated) notifyInsert(c, job.queue()); return job; } @@ -809,7 +887,7 @@ static RiverException databaseError(String action, SQLException error) { } Job typed(Job job, JobType type) { - return job.mapArgs(ignored -> Json.decode(plugin.decode(job), type.argsType())); + return job.mapArgs(ignored -> Json.decode(pilot.decode(job), type.argsType())); } Job read(ResultSet rows) throws SQLException { diff --git a/java/river/src/main/java/com/riverqueue/Extension.java b/java/river/src/main/java/com/riverqueue/Extension.java deleted file mode 100644 index a1b54bef4..000000000 --- a/java/river/src/main/java/com/riverqueue/Extension.java +++ /dev/null @@ -1,27 +0,0 @@ -package com.riverqueue; - -import java.sql.Connection; -import java.util.concurrent.Callable; - -/** Transaction-scoped insert hooks and attempt-scoped worker middleware. */ -public interface Extension { - default void beforeInsert( - Connection connection, JobType type, Object args, InsertOptions options) - throws Exception {} - - default void beforeWork(WorkContext context) throws Exception {} - - default void afterWork(WorkContext context, Throwable failure) throws Exception {} - - default void afterClaim(Job job) throws Exception {} - - default T insert(Connection connection, Callable next) throws Exception { - return next.call(); - } - - default void periodicStarted() {} - - default void work(WorkContext context, WorkContext.Step next) throws Exception { - next.run(); - } -} diff --git a/java/river/src/main/java/com/riverqueue/Hook.java b/java/river/src/main/java/com/riverqueue/Hook.java new file mode 100644 index 000000000..19b0b0fac --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/Hook.java @@ -0,0 +1,10 @@ +package com.riverqueue; + +/** + * A lifecycle hook registered with {@link Client#withHooks(Hook...)}. Implement one or more of + * {@link HookInsertBegin}, {@link HookPeriodicJobsStart}, {@link HookWorkBegin}, and {@link + * HookWorkEnd}. Hooks run in registration order and may be called concurrently. Use {@link + * Client#withPlugins(Plugin...)} to also activate any middleware interfaces implemented by the same + * instance. + */ +public interface Hook extends Plugin {} diff --git a/java/river/src/main/java/com/riverqueue/HookInsertBegin.java b/java/river/src/main/java/com/riverqueue/HookInsertBegin.java new file mode 100644 index 000000000..e0b6e67dc --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/HookInsertBegin.java @@ -0,0 +1,15 @@ +package com.riverqueue; + +import java.sql.Connection; + +/** A hook run for each job inside insertion middleware, before the database write. */ +@FunctionalInterface +public interface HookInsertBegin extends Hook { + /** + * Runs in the insertion's transaction. Throwing fails and rolls back the insertion, including + * application writes made using this connection. Options are the caller's overrides, before + * merging with job-type defaults. + */ + void insertBegin(Connection connection, JobType type, Object args, InsertOptions options) + throws Exception; +} diff --git a/java/river/src/main/java/com/riverqueue/HookPeriodicJobsStart.java b/java/river/src/main/java/com/riverqueue/HookPeriodicJobsStart.java new file mode 100644 index 000000000..0988a4f8a --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/HookPeriodicJobsStart.java @@ -0,0 +1,11 @@ +package com.riverqueue; + +/** A hook run when periodic job enqueuing starts on a newly elected leader. */ +@FunctionalInterface +public interface HookPeriodicJobsStart extends Hook { + /** + * Runs once per leadership term after election commits. Throwing prevents maintenance from + * starting and retries initialization on the next leadership check. + */ + void start() throws Exception; +} diff --git a/java/river/src/main/java/com/riverqueue/HookWorkBegin.java b/java/river/src/main/java/com/riverqueue/HookWorkBegin.java new file mode 100644 index 000000000..90d00bd80 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/HookWorkBegin.java @@ -0,0 +1,11 @@ +package com.riverqueue; + +/** A hook run inside worker middleware, before the handler. */ +@FunctionalInterface +public interface HookWorkBegin extends Hook { + /** + * Runs with decoded arguments. Throwing fails the attempt without running the handler or work-end + * hooks. + */ + void workBegin(WorkContext context) throws Exception; +} diff --git a/java/river/src/main/java/com/riverqueue/HookWorkEnd.java b/java/river/src/main/java/com/riverqueue/HookWorkEnd.java new file mode 100644 index 000000000..5d00b1e74 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/HookWorkEnd.java @@ -0,0 +1,13 @@ +package com.riverqueue; + +/** A hook run inside worker middleware after the handler returns or throws. */ +@FunctionalInterface +public interface HookWorkEnd extends Hook { + /** + * Receives the handler's failure, or {@code null} on success. Return it unchanged to preserve the + * outcome, another failure to replace it, or {@code null} to report success. Each hook receives + * the previous hook's result, including control exceptions for cancellation and snoozing. + * Throwing fails the attempt and skips remaining work-end hooks. + */ + Throwable workEnd(WorkContext context, Throwable failure) throws Exception; +} diff --git a/java/river/src/main/java/com/riverqueue/JobInsertMiddleware.java b/java/river/src/main/java/com/riverqueue/JobInsertMiddleware.java new file mode 100644 index 000000000..652e5f118 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/JobInsertMiddleware.java @@ -0,0 +1,21 @@ +package com.riverqueue; + +import java.sql.Connection; +import java.util.List; +import java.util.concurrent.Callable; + +/** Middleware around an insertion, including its hooks and database writes. */ +@FunctionalInterface +public interface JobInsertMiddleware extends Middleware { + /** + * Wraps a whole batch in its transaction; a single insertion is a one-element batch. Submissions + * are an immutable list of the caller's requests, before merging job-type defaults. Call {@code + * next.call()} exactly once and return its results in input order. Throwing, even after {@code + * next.call()}, rolls back the insertion and application writes made using this connection. + */ + List insertMany( + Connection connection, + List> submissions, + Callable> next) + throws Exception; +} diff --git a/java/river/src/main/java/com/riverqueue/Middleware.java b/java/river/src/main/java/com/riverqueue/Middleware.java new file mode 100644 index 000000000..9232571a3 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/Middleware.java @@ -0,0 +1,9 @@ +package com.riverqueue; + +/** + * Middleware registered with {@link Client#withMiddleware(Middleware...)}. Implement {@link + * JobInsertMiddleware}, {@link WorkerMiddleware}, or both. The first registered middleware is + * outermost. Instances may be called concurrently. Use {@link Client#withPlugins(Plugin...)} to + * also activate any hook interfaces implemented by the same instance. + */ +public interface Middleware extends Plugin {} diff --git a/java/river/src/main/java/com/riverqueue/Pilot.java b/java/river/src/main/java/com/riverqueue/Pilot.java new file mode 100644 index 000000000..89596f37f --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/Pilot.java @@ -0,0 +1,110 @@ +package com.riverqueue; + +import java.sql.Connection; +import java.time.Instant; +import java.util.List; +import java.util.Set; +import tools.jackson.databind.JsonNode; + +/** + * Internal transactional storage seam for River Pro. This interface is released in lockstep with + * River Pro and is not a stable application extension API; applications should use {@link Plugin}. + */ +public interface Pilot { + default void afterInsert(Connection connection, Client.Driver driver, Job job) + throws Exception {} + + default void afterDelete(Connection connection, Client.Driver driver, Job job) + throws Exception {} + + default void afterStateChange(Connection connection, Client.Driver driver, Job job) + throws Exception {} + + default void afterAttempt(Connection connection, Client.Driver driver, Job original) + throws Exception {} + + default void afterPeerClaim( + Connection connection, Client.Driver driver, List jobs) throws Exception {} + + default List claim( + Connection connection, + Client.Driver driver, + Claim claim, + Client.Transaction> next) + throws Exception { + return next.run(connection); + } + + default boolean clean(Client river, Instant now) { + return false; + } + + default boolean clean( + Client river, + Instant now, + java.time.Duration cancelled, + java.time.Duration completed, + java.time.Duration discarded) { + return clean(river, now); + } + + default JsonNode decode(Job job) { + return job.args(); + } + + default void maintain(Client river, Instant now) {} + + default Insert prepare(Connection connection, Client.Driver driver, Insert insert) + throws Exception { + return insert; + } + + default void producer(Connection connection, Client.Driver driver, Producer producer) + throws Exception {} + + default void leadershipStarted() {} + + default int workerLimit(String queue, int configured) { + return configured; + } + + default void starting(boolean leadership) {} + + default void starting( + boolean leadership, + java.time.Duration cancelled, + java.time.Duration completed, + java.time.Duration discarded) { + starting(leadership); + } + + default boolean rescue(Client river, Rescue rescue) { + return false; + } + + record Rescue( + Instant now, + Instant horizon, + Set activeIds, + Set kinds, + boolean timeoutDisabled, + RetryPolicy retryPolicy) {} + + record Claim( + String clientId, + String queue, + int limit, + Instant now, + Set kinds, + boolean fetchOnlyKnownKinds) {} + + record Insert(JobType type, String args, InsertOptions options) {} + + record Producer( + String clientId, + String queue, + int maxWorkers, + List> active, + boolean paused, + Instant now) {} +} diff --git a/java/river/src/main/java/com/riverqueue/Plugin.java b/java/river/src/main/java/com/riverqueue/Plugin.java index c8662bf9d..5110cb34c 100644 --- a/java/river/src/main/java/com/riverqueue/Plugin.java +++ b/java/river/src/main/java/com/riverqueue/Plugin.java @@ -1,111 +1,12 @@ package com.riverqueue; -import java.sql.Connection; -import java.time.Instant; -import java.util.List; -import java.util.Set; -import tools.jackson.databind.JsonNode; - /** - * Internal transactional storage seam for River Pro. This interface is released in lockstep with - * River Pro and is not a stable application extension API; applications should use {@link - * Extension}. + * A shared type for hooks and middleware installed with {@link Client#withPlugins(Plugin...)}. + * Implement any combination of the operation-specific {@link Hook} and {@link Middleware} + * interfaces; they already extend this interface, so no additional methods are required. + * + *

Registration activates every hook and middleware interface the instance implements. Plugins + * run in registration order, before separately registered hooks and middleware. The first + * middleware is outermost. Instances may be invoked concurrently. */ -public interface Plugin { - default void afterInsert(Connection connection, Client.Driver driver, Job job) - throws Exception {} - - default void afterDelete(Connection connection, Client.Driver driver, Job job) - throws Exception {} - - default void afterStateChange(Connection connection, Client.Driver driver, Job job) - throws Exception {} - - default void afterAttempt(Connection connection, Client.Driver driver, Job original) - throws Exception {} - - default void afterPeerClaim( - Connection connection, Client.Driver driver, List jobs) throws Exception {} - - default List claim( - Connection connection, - Client.Driver driver, - Claim claim, - Client.Transaction> next) - throws Exception { - return next.run(connection); - } - - default boolean clean(Client river, Instant now) { - return false; - } - - default boolean clean( - Client river, - Instant now, - java.time.Duration cancelled, - java.time.Duration completed, - java.time.Duration discarded) { - return clean(river, now); - } - - default JsonNode decode(Job job) { - return job.args(); - } - - default void maintain(Client river, Instant now) {} - - default Insert prepare(Connection connection, Client.Driver driver, Insert insert) - throws Exception { - return insert; - } - - default void producer(Connection connection, Client.Driver driver, Producer producer) - throws Exception {} - - default void leadershipStarted() {} - - default int workerLimit(String queue, int configured) { - return configured; - } - - default void starting(boolean leadership) {} - - default void starting( - boolean leadership, - java.time.Duration cancelled, - java.time.Duration completed, - java.time.Duration discarded) { - starting(leadership); - } - - default boolean rescue(Client river, Rescue rescue) { - return false; - } - - record Rescue( - Instant now, - Instant horizon, - Set activeIds, - Set kinds, - boolean timeoutDisabled, - RetryPolicy retryPolicy) {} - - record Claim( - String clientId, - String queue, - int limit, - Instant now, - Set kinds, - boolean fetchOnlyKnownKinds) {} - - record Insert(JobType type, String args, InsertOptions options) {} - - record Producer( - String clientId, - String queue, - int maxWorkers, - List> active, - boolean paused, - Instant now) {} -} +public interface Plugin {} diff --git a/java/river/src/main/java/com/riverqueue/WorkerMiddleware.java b/java/river/src/main/java/com/riverqueue/WorkerMiddleware.java new file mode 100644 index 000000000..f7fc43fd5 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/WorkerMiddleware.java @@ -0,0 +1,11 @@ +package com.riverqueue; + +/** Middleware around one job attempt, including its work hooks and handler. */ +@FunctionalInterface +public interface WorkerMiddleware extends Middleware { + /** + * Runs with decoded arguments. Call {@code next.run()} once to run the inner middleware, hooks, + * and handler. Throwing fails the attempt, even if the inner work succeeded. + */ + void work(WorkContext context, WorkContext.Step next) throws Exception; +} diff --git a/java/river/src/main/java/com/riverqueue/Workers.java b/java/river/src/main/java/com/riverqueue/Workers.java index 4da311738..45dd72c4b 100644 --- a/java/river/src/main/java/com/riverqueue/Workers.java +++ b/java/river/src/main/java/com/riverqueue/Workers.java @@ -242,7 +242,7 @@ private void dispatch() { for (var queue : queues.entrySet()) { if (drainingQueues.contains(queue.getKey())) continue; int available = - river.plugin().workerLimit(queue.getKey(), queue.getValue()) + river.pilot().workerLimit(queue.getKey(), queue.getValue()) - (int) active.values().stream() .filter(c -> c.job().queue().equals(queue.getKey())) @@ -335,11 +335,11 @@ private void claim(String queue, int count) { river.transaction( c -> river - .plugin() + .pilot() .claim( c, river.driver(), - new Plugin.Claim( + new Pilot.Claim( id, queue, count, @@ -401,22 +401,24 @@ private void launch(Client.Decoded decoded, Registration registration) { row, context, () -> { - for (var extension : river.extensions()) extension.afterClaim(row); workMiddleware( context, 0, () -> { + for (var hook : river.hooks()) + if (hook instanceof HookWorkBegin workHook) workHook.workBegin(context); Throwable workFailure = null; try { - for (var extension : river.extensions()) extension.beforeWork(context); registration.handler.work(context); } catch (Exception | Error e) { workFailure = e; - throw e; - } finally { - for (var extension : river.extensions()) - extension.afterWork(context, workFailure); } + for (var hook : river.hooks()) + if (hook instanceof HookWorkEnd workHook) + workFailure = workHook.workEnd(context, workFailure); + if (workFailure instanceof Exception exception) throw exception; + if (workFailure instanceof Error error) throw error; + if (workFailure != null) throw new RuntimeException(workFailure); }); }); } @@ -482,9 +484,12 @@ private void launch(Job row, WorkContext context, WorkContext.Step private void workMiddleware(WorkContext context, int index, WorkContext.Step next) throws Exception { - if (index == river.extensions().size()) next.run(); + if (index == river.workerMiddleware().size()) next.run(); else - river.extensions().get(index).work(context, () -> workMiddleware(context, index + 1, next)); + river + .workerMiddleware() + .get(index) + .work(context, () -> workMiddleware(context, index + 1, next)); } List> claimPeers( @@ -494,7 +499,7 @@ List> claimPeers( river.transaction( connection -> { var claimed = operation.run(connection); - river.plugin().afterPeerClaim(connection, river.driver(), claimed); + river.pilot().afterPeerClaim(connection, river.driver(), claimed); return claimed; }); // Register every committed claim before decoding any arguments. A decoding failure must still @@ -677,9 +682,9 @@ private void completeBatches() { } } for (var job : rows) - if (job != null) river.plugin().afterStateChange(c, river.driver(), job); + if (job != null) river.pilot().afterStateChange(c, river.driver(), job); for (var completion : batch) - river.plugin().afterAttempt(c, river.driver(), completion.original); + river.pilot().afterAttempt(c, river.driver(), completion.original); return rows; }); for (int i = 0; i < batch.size(); i++) batch.get(i).done.complete(results.get(i)); @@ -706,11 +711,11 @@ private void heartbeat(String queue) { statement.executeUpdate(); } river - .plugin() + .pilot() .producer( c, river.driver(), - new Plugin.Producer( + new Pilot.Producer( id, queue, queues.getOrDefault(queue, config.queues.getOrDefault(queue, 1)), @@ -893,7 +898,7 @@ private void elect() { } } - private void maintain() { + private void maintain() throws Exception { leadershipLock.lock(); try { if (resigningTerm != null) resign(); @@ -957,8 +962,9 @@ private void maintain() { if (!running.get()) return; if (!elected.equals(initializedTerm)) { periodicNext.clear(); - river.plugin().leadershipStarted(); - river.extensions().forEach(Extension::periodicStarted); + river.pilot().leadershipStarted(); + for (var hook : river.hooks()) + if (hook instanceof HookPeriodicJobsStart periodicHook) periodicHook.start(); initializedTerm = elected; } Instant term = elected; @@ -1019,7 +1025,7 @@ private void services(Instant now, Instant term) throws SQLException { if (!ownsTerm(term)) return; schedule(now); if (!ownsTerm(term)) return; - river.plugin().maintain(river, now); + river.pilot().maintain(river, now); if (!ownsTerm(term)) return; rescue(now, term); if (!ownsTerm(term)) return; @@ -1055,7 +1061,7 @@ boolean active(long now) { private void clean(Instant now) { if (river - .plugin() + .pilot() .clean( river, now, @@ -1188,10 +1194,10 @@ private record Scheduled(long id, byte[] key) {} private void rescue(Instant now, Instant term) { if (river - .plugin() + .pilot() .rescue( river, - new Plugin.Rescue( + new Pilot.Rescue( now, now.minus(config.rescueAfter), java.util.Set.copyOf(active.keySet()), @@ -1645,7 +1651,7 @@ public Workers start() { if (config.rescueAfter.compareTo(jobTimeout) < 0) throw new IllegalArgumentException("rescueAfter must not be less than jobTimeout"); river - .plugin() + .pilot() .starting(leadership, cancelledRetention, completedRetention, discardedRetention); return new Workers(config); } diff --git a/java/river/src/main/java/com/riverqueue/package-info.java b/java/river/src/main/java/com/riverqueue/package-info.java index f36b30121..bfc4db07b 100644 --- a/java/river/src/main/java/com/riverqueue/package-info.java +++ b/java/river/src/main/java/com/riverqueue/package-info.java @@ -14,7 +14,7 @@ * corresponding value, such as a job's finalization time. Missing jobs throw an exception with code * {@code NOT_FOUND}. * - *

{@link com.riverqueue.Plugin}, {@link com.riverqueue.Client.Driver}, and members explicitly + *

{@link com.riverqueue.Pilot}, {@link com.riverqueue.Client.Driver}, and members explicitly * marked internal are integration seams for the matching Pro release, not stable application APIs. */ package com.riverqueue; diff --git a/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java b/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java index 0a1e178cd..f3300790a 100644 --- a/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java +++ b/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java @@ -103,8 +103,8 @@ void hookFailureRollsBackWholeBatch() throws Exception { var first = base.insert(TYPE, "first").job(); var second = base.insert(TYPE, "second").job(); var client = - base.withPlugin( - new Plugin() { + base.withPilot( + new Pilot() { @Override public void afterDelete( Connection connection, Client.Driver driver, Job job) { @@ -136,8 +136,8 @@ void ignoresJobsChangedAfterListing(String change) { var last = base.insert(TYPE, "last").job(); var deleted = new ArrayList(); var client = - base.withPlugin( - new Plugin() { + base.withPilot( + new Pilot() { @Override public void afterDelete( Connection connection, Client.Driver driver, Job job) { @@ -219,8 +219,8 @@ void preservesJobsThatNoLongerMatchFilters(String change) { .build(); var deleted = new ArrayList(); var client = - base.withPlugin( - new Plugin() { + base.withPilot( + new Pilot() { @Override public void afterDelete( Connection connection, Client.Driver driver, Job job) { diff --git a/java/river/src/test/java/com/riverqueue/ClientTest.java b/java/river/src/test/java/com/riverqueue/ClientTest.java index fadbe9574..882b4d3cb 100644 --- a/java/river/src/test/java/com/riverqueue/ClientTest.java +++ b/java/river/src/test/java/com/riverqueue/ClientTest.java @@ -209,10 +209,10 @@ void failingHookRollsBackItsApplicationWrites() { return null; }); var river = - base.withExtension( - new Extension() { + base.withHooks( + new HookInsertBegin() { @Override - public void beforeInsert( + public void insertBegin( Connection connection, JobType type, Object args, InsertOptions options) { try (var statement = connection.createStatement()) { statement.executeUpdate( @@ -437,10 +437,10 @@ void typedGetAndMixedBatchRetainArgumentKinds() throws Exception { void checkedHookFailureRollsBackInsideCallerTransaction() throws Exception { var base = database(); var river = - base.withExtension( - new Extension() { + base.withHooks( + new HookInsertBegin() { @Override - public void beforeInsert( + public void insertBegin( Connection connection, JobType type, Object args, InsertOptions options) throws java.sql.SQLException { try (var statement = connection.createStatement()) { @@ -631,8 +631,8 @@ void stopAfterTimeoutStillWaitsForActiveWorker() throws Exception { var releaseAcknowledgement = new CountDownLatch(1); var river = database() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterAttempt( Connection connection, diff --git a/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java b/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java new file mode 100644 index 000000000..727131840 --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java @@ -0,0 +1,312 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.io.IOException; +import java.nio.file.Path; +import java.sql.Connection; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +class HookMiddlewareTest { + private static final JobType TYPE = JobType.of("hook_test", String.class); + @RegisterExtension final TestDatabase databases = new TestDatabase(); + @TempDir Path directory; + + private Client client() { + return new Client(databases.open(directory.resolve("river.db"))); + } + + @ParameterizedTest + @ValueSource(strings = {"single", "bulk", "mixed", "reinsert"}) + void insertHooksAndMiddlewareRunInOrder(String operation) { + var base = client(); + var trace = new ArrayList(); + var connectionSeen = new AtomicReference(); + var submissionsSeen = new AtomicReference>>(); + var resultsSeen = new AtomicReference>(); + class Instrumentation implements HookInsertBegin, JobInsertMiddleware { + private final String name; + + Instrumentation(String name) { + this.name = name; + } + + @Override + public void insertBegin( + Connection connection, JobType type, Object args, InsertOptions options) { + assertSame(connectionSeen.get(), connection); + trace.add(name + ":hook:" + Json.tree(args).asString()); + } + + @Override + public List insertMany( + Connection connection, + List> submissions, + Callable> next) + throws Exception { + if (connectionSeen.compareAndSet(null, connection)) { + submissionsSeen.set(submissions); + assertThrows(UnsupportedOperationException.class, submissions::clear); + } else { + assertSame(connectionSeen.get(), connection); + assertSame(submissionsSeen.get(), submissions); + } + trace.add(name + ":before"); + var result = next.call(); + resultsSeen.set(result); + trace.add(name + ":after"); + return result; + } + } + var first = new Instrumentation("first"); + var second = new Instrumentation("second"); + var river = + base.withHooks(first) + .withMiddleware(first) + .withHooks(second) + .withMiddleware(second) + .withPilot(new Pilot() {}); + List results = + switch (operation) { + case "single" -> List.of(river.insert(TYPE, "one")); + case "bulk" -> river.insertMany(TYPE, List.of("one", "two")); + case "mixed" -> + river.insertMany( + List.of( + TYPE.submission("one"), + JobType.of("other_kind", String.class).submission("two"))); + case "reinsert" -> { + var original = base.insert(TYPE, "one").job(); + assertTrue(trace.isEmpty(), "Registration must leave the original client unchanged"); + var reinserted = + river.transaction(connection -> river.driver().reinsert(connection, original)); + yield List.of(reinserted); + } + default -> throw new IllegalArgumentException(operation); + }; + var expected = + new ArrayList<>( + List.of("first:before", "second:before", "first:hook:one", "second:hook:one")); + if (results.size() == 2) expected.addAll(List.of("first:hook:two", "second:hook:two")); + expected.addAll(List.of("second:after", "first:after")); + assertEquals(expected, trace); + assertEquals(results, resultsSeen.get()); + assertEquals( + results.stream().map(result -> result.job().kind()).toList(), + submissionsSeen.get().stream().map(submission -> submission.type().kind()).toList()); + assertEquals( + results.stream().map(result -> result.job().args().asString()).toList(), + submissionsSeen.get().stream() + .map(submission -> Json.tree(submission.args()).asString()) + .toList()); + } + + @ParameterizedTest + @CsvSource({ + "single,false", + "single,true", + "bulk,false", + "bulk,true", + "mixed,false", + "mixed,true" + }) + void insertMiddlewareFailureRollsBackWrites(String operation, boolean callerTransaction) + throws Exception { + var base = client(); + var failure = new IOException("after insertion"); + var river = + base.withMiddleware( + (JobInsertMiddleware) + (connection, submissions, next) -> { + next.call(); + base.insert(connection, TYPE, "middleware write"); + throw failure; + }); + try (var connection = base.database().connection()) { + connection.setAutoCommit(false); + if (callerTransaction) base.insert(connection, TYPE, "keep"); + var error = + assertThrows( + RiverException.class, + () -> { + switch (operation) { + case "single" -> { + if (callerTransaction) river.insert(connection, TYPE, "reject"); + else river.insert(TYPE, "reject"); + } + case "bulk" -> { + if (callerTransaction) + river.insertMany(connection, TYPE, List.of("reject", "also reject")); + else river.insertMany(TYPE, List.of("reject", "also reject")); + } + case "mixed" -> { + var submissions = + List.of(TYPE.submission("reject"), TYPE.submission("also reject")); + if (callerTransaction) river.insertMany(connection, submissions); + else river.insertMany(submissions); + } + default -> throw new IllegalArgumentException(operation); + } + }); + assertSame(failure, error.getCause()); + connection.commit(); + } + assertEquals( + callerTransaction ? List.of("keep") : List.of(), + base.list(JobQuery.all()).jobs().stream().map(job -> job.args().asString()).toList()); + } + + @Test + void registrationRejectsNullEntries() { + var river = client(); + assertThrows(NullPointerException.class, () -> river.withHooks((Hook) null)); + assertThrows(NullPointerException.class, () -> river.withMiddleware((Middleware) null)); + } + + @Test + void workBeginFailureSkipsHandlerAndEndHooks() throws Exception { + var trace = new ArrayList(); + var river = + client() + .withHooks( + (HookWorkBegin) + context -> { + trace.add("begin"); + throw new IOException("begin failed"); + }, + (HookWorkEnd) + (context, failure) -> { + trace.add("end"); + return failure; + }) + .withMiddleware( + (WorkerMiddleware) + (context, next) -> { + trace.add("before"); + try { + next.run(); + } finally { + trace.add("after"); + } + }); + var completed = new CompletableFuture(); + try (var workers = workers(river).add(TYPE, context -> trace.add("handler")).start(); + var subscription = workers.subscribe(completed::complete)) { + river.insert(TYPE, "one", InsertOptions.builder().maxAttempts(1).build()); + assertEquals(Job.State.DISCARDED, completed.get(5, TimeUnit.SECONDS).job().state()); + } + assertEquals(List.of("before", "begin", "after"), trace); + } + + @ParameterizedTest + @CsvSource({ + "success,preserve", + "failure,preserve", + "error,preserve", + "cancel,preserve", + "failure,clear", + "error,clear", + "success,replace", + "failure,replace", + "success,throw" + }) + void workEndHooksChainOutcomesInsideMiddleware(String outcome, String action) throws Exception { + var trace = new ArrayList(); + var failure = new IOException("handler failed"); + var fatal = new AssertionError("handler error"); + var replacement = new IOException("replacement"); + var observed = new AtomicReference(); + var river = + client() + .withHooks( + (HookWorkBegin) context -> trace.add("begin"), + (HookWorkEnd) + (context, error) -> { + trace.add("end:first"); + switch (outcome) { + case "success" -> assertNull(error); + case "failure" -> assertSame(failure, error); + case "error" -> assertSame(fatal, error); + case "cancel" -> assertInstanceOf(WorkContext.Control.class, error); + default -> throw new IllegalArgumentException(outcome); + } + return switch (action) { + case "clear" -> null; + case "replace" -> replacement; + case "throw" -> throw replacement; + default -> error; + }; + }, + (HookWorkEnd) + (context, error) -> { + trace.add("end:second"); + observed.set(error); + return error; + }); + for (var name : List.of("first", "second")) { + river = + river.withMiddleware( + (WorkerMiddleware) + (context, next) -> { + trace.add(name + ":before"); + try { + next.run(); + } finally { + trace.add(name + ":after"); + } + }); + } + var completed = new CompletableFuture(); + try (var workers = + workers(river) + .add( + TYPE, + context -> { + trace.add("handler"); + switch (outcome) { + case "failure" -> throw failure; + case "error" -> throw fatal; + case "cancel" -> context.cancel("cancelled"); + default -> {} + } + }) + .start(); + var subscription = workers.subscribe(completed::complete)) { + river.insert(TYPE, "one", InsertOptions.builder().maxAttempts(1).build()); + var expected = + action.equals("clear") || (outcome.equals("success") && action.equals("preserve")) + ? Job.State.COMPLETED + : outcome.equals("cancel") ? Job.State.CANCELLED : Job.State.DISCARDED; + assertEquals(expected, completed.get(5, TimeUnit.SECONDS).job().state()); + } + var expected = + new ArrayList<>(List.of("first:before", "second:before", "begin", "handler", "end:first")); + if (!action.equals("throw")) expected.add("end:second"); + expected.addAll(List.of("second:after", "first:after")); + assertEquals(expected, trace); + if (action.equals("replace")) assertSame(replacement, observed.get()); + if (action.equals("clear")) assertNull(observed.get()); + } + + private Workers.Builder workers(Client river) { + return river + .workers() + .queue("default", 1) + .leadership(false) + .pollOnly(true) + .pollInterval(Duration.ofMillis(10)) + .stopTimeout(Duration.ofSeconds(5)); + } +} diff --git a/java/river/src/test/java/com/riverqueue/LeadershipTest.java b/java/river/src/test/java/com/riverqueue/LeadershipTest.java index 6e4bfcc1e..b66b7209e 100644 --- a/java/river/src/test/java/com/riverqueue/LeadershipTest.java +++ b/java/river/src/test/java/com/riverqueue/LeadershipTest.java @@ -67,10 +67,10 @@ void failedElectionCommitDoesNotPublishLeadership() throws Exception { var reported = new CompletableFuture(); var client = new Client(failLeadershipCommit(database, failCommit)) - .withExtension( - new Extension() { + .withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { started.countDown(); } }); @@ -190,10 +190,10 @@ void reacquisitionStartsANewTerm() throws Exception { var starts = new LinkedBlockingQueue(); var errors = new LinkedBlockingQueue(); client = - client.withExtension( - new Extension() { + client.withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { starts.add(true); } }); @@ -252,10 +252,10 @@ void resignationWakesPeersOnBothDatabases() throws Exception { try (first) { assertTrue(firstCheck.await(5, TimeUnit.SECONDS)); var follower = - client.withExtension( - new Extension() { + client.withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { elected.countDown(); } }); @@ -297,8 +297,8 @@ void resignedTermDoesNotContinueMaintenance() throws Exception { var rescues = new AtomicInteger(); var errors = new LinkedBlockingQueue(); var blocked = - base.withPlugin( - new Plugin() { + base.withPilot( + new Pilot() { @Override public void maintain(Client client, Instant now) { entered.countDown(); @@ -330,10 +330,10 @@ public boolean rescue(Client client, Rescue request) { assertTrue(entered.await(5, TimeUnit.SECONDS)); var elected = new CountDownLatch(1); var peer = - base.withExtension( - new Extension() { + base.withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { elected.countDown(); } }); @@ -369,10 +369,10 @@ void startupHooksAreRetriedAfterFailure() throws Exception { var errors = new LinkedBlockingQueue(); var client = new Client(databases.open(directory.resolve("river.db"))) - .withExtension( - new Extension() { + .withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { if (attempts.incrementAndGet() == 1) throw new IllegalStateException("startup failed"); started.countDown(); @@ -406,8 +406,8 @@ void stopRetriesFailedFinalization(String failure) throws Exception { var checked = new CountDownLatch(1); var client = new Client(failLeadershipCommit(database, failCommit)) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void producer( Connection connection, Client.Driver driver, Producer producer) diff --git a/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java b/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java index a53a157a0..002a79db0 100644 --- a/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java +++ b/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java @@ -76,8 +76,8 @@ void leaderCleansNotificationsEvenWhenPluginOwnsJobCleaning(boolean pluginOwnsCl var cleaned = new CompletableFuture>(); var errors = new LinkedBlockingQueue(); var client = - base.withPlugin( - new Plugin() { + base.withPilot( + new Pilot() { @Override public boolean clean(Client river, Instant now) { // The next maintenance stage observes the notification cleanup's committed result. diff --git a/java/river/src/test/java/com/riverqueue/NotificationTest.java b/java/river/src/test/java/com/riverqueue/NotificationTest.java index 0cf20201e..4e443cb92 100644 --- a/java/river/src/test/java/com/riverqueue/NotificationTest.java +++ b/java/river/src/test/java/com/riverqueue/NotificationTest.java @@ -51,8 +51,8 @@ void sqliteListenerRecoversFromAnInitialDatabaseFailure() throws Exception { var initiallyFetched = new CompletableFuture(); var client = new Client(new Database(source, Database.Dialect.SQLITE)) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public List claim( Connection connection, diff --git a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java index 0f5698572..21992b839 100644 --- a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java +++ b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java @@ -37,8 +37,8 @@ void cancellationDuringArgumentDecodingReachesTheTypedPeer( var database = databases.open(directory.resolve("river.db")); var client = new Client(database) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public JsonNode decode(Job job) { if (job.queue().equals("peers")) { diff --git a/java/river/src/test/java/com/riverqueue/PluginTest.java b/java/river/src/test/java/com/riverqueue/PluginTest.java new file mode 100644 index 000000000..0c4396744 --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/PluginTest.java @@ -0,0 +1,237 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.nio.file.Path; +import java.sql.Connection; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import tools.jackson.databind.JsonNode; + +class PluginTest { + private static final JobType TYPE = JobType.of("plugin_test", String.class); + @RegisterExtension final TestDatabase databases = new TestDatabase(); + @TempDir Path directory; + + private Client client() { + return new Client(databases.open(directory.resolve("river.db"))); + } + + @ParameterizedTest + @ValueSource(strings = {"hooks", "middleware"}) + void explicitRegistrationActivatesOnlyItsOwnInterfaces(String registration) { + var trace = new CopyOnWriteArrayList(); + var plugin = new RecordingPlugin("explicit", trace); + var base = client(); + var river = registration.equals("hooks") ? base.withHooks(plugin) : base.withMiddleware(plugin); + river.insert(TYPE, "one"); + assertEquals( + registration.equals("hooks") + ? List.of("explicit:insert_begin") + : List.of("explicit:insert_before", "explicit:insert_after"), + trace); + } + + @Test + void pluginRegistrationCopiesInputsAndPreservesThePilot() { + var trace = new CopyOnWriteArrayList(); + var base = + client() + .withPilot( + new Pilot() { + @Override + public void afterInsert( + Connection connection, Client.Driver driver, Job job) { + trace.add("storage"); + } + }); + Plugin[] plugins = {(HookInsertBegin) (connection, type, args, options) -> trace.add("first")}; + var first = base.withPlugins(plugins); + plugins[0] = (HookInsertBegin) (connection, type, args, options) -> fail("Replaced input"); + var second = + first.withPlugins( + (HookInsertBegin) (connection, type, args, options) -> trace.add("second"), + new Plugin() {}); + base.insert(TYPE, "base"); + assertEquals(List.of("storage"), trace); + trace.clear(); + first.insert(TYPE, "first"); + assertEquals(List.of("first", "storage"), trace); + trace.clear(); + second.insert(TYPE, "second"); + assertEquals(List.of("first", "second", "storage"), trace); + } + + @Test + void pluginRegistrationRetainsDuplicates() { + var trace = new CopyOnWriteArrayList(); + var plugin = new RecordingPlugin("repeat", trace); + var river = client().withPlugins(plugin, plugin).withHooks(plugin).withMiddleware(plugin); + river.insert(TYPE, "one"); + assertEquals( + List.of( + "repeat:insert_before", + "repeat:insert_before", + "repeat:insert_before", + "repeat:insert_begin", + "repeat:insert_begin", + "repeat:insert_begin", + "repeat:insert_after", + "repeat:insert_after", + "repeat:insert_after"), + trace); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void pluginsActivateAllJobInterfacesBeforeExplicitRegistrations(boolean pluginsFirst) + throws Exception { + var trace = new CopyOnWriteArrayList(); + var first = new RecordingPlugin("first", trace); + var second = new RecordingPlugin("second", trace); + var explicit = new RecordingPlugin("explicit", trace); + var base = client(); + var river = + (pluginsFirst + ? base.withPlugins(first, second).withHooks(explicit).withMiddleware(explicit) + : base.withHooks(explicit) + .withMiddleware(explicit) + .withPlugins(first) + .withPlugins(second)) + .withPilot(new Pilot() {}); + var completed = new CompletableFuture(); + try (var workers = + river + .workers() + .queue("default", 1) + .leadership(false) + .pollOnly(true) + .pollInterval(Duration.ofMillis(10)) + .stopTimeout(Duration.ofSeconds(5)) + .add(TYPE, context -> trace.add("handler")) + .start(); + var subscription = workers.subscribe(completed::complete)) { + river.insert(TYPE, "one"); + assertEquals(Job.State.COMPLETED, completed.get(5, TimeUnit.SECONDS).job().state()); + } + assertEquals( + List.of( + "first:insert_before", + "second:insert_before", + "explicit:insert_before", + "first:insert_begin", + "second:insert_begin", + "explicit:insert_begin", + "explicit:insert_after", + "second:insert_after", + "first:insert_after", + "first:work_before", + "second:work_before", + "explicit:work_before", + "first:work_begin", + "second:work_begin", + "explicit:work_begin", + "handler", + "first:work_end", + "second:work_end", + "explicit:work_end", + "explicit:work_after", + "second:work_after", + "first:work_after"), + trace); + } + + @Test + void pluginsIncludePeriodicStartHooks() throws Exception { + var started = new LinkedBlockingQueue(); + Plugin first = (HookPeriodicJobsStart) () -> started.add("first"); + Plugin second = (HookPeriodicJobsStart) () -> started.add("second"); + var river = + client() + .withHooks((HookPeriodicJobsStart) () -> started.add("explicit")) + .withPlugins(first, second); + try (var workers = + river + .workers() + .queue("default", 1) + .pollOnly(true) + .serviceInterval(Duration.ofHours(1)) + .stopTimeout(Duration.ofSeconds(5)) + .add(TYPE, context -> {}) + .start()) { + assertEquals("first", started.poll(5, TimeUnit.SECONDS)); + assertEquals("second", started.poll(5, TimeUnit.SECONDS)); + assertEquals("explicit", started.poll(5, TimeUnit.SECONDS)); + assertTrue(workers.isLeader()); + } + assertTrue(started.isEmpty()); + } + + @Test + void registrationRejectsNullPlugins() { + var river = client(); + assertThrows(NullPointerException.class, () -> river.withPlugins((Plugin[]) null)); + assertThrows(NullPointerException.class, () -> river.withPlugins(new Plugin() {}, null)); + } + + private static final class RecordingPlugin + implements HookInsertBegin, + HookWorkBegin, + HookWorkEnd, + JobInsertMiddleware, + WorkerMiddleware { + private final String name; + private final List trace; + + RecordingPlugin(String name, List trace) { + this.name = name; + this.trace = trace; + } + + @Override + public void insertBegin( + Connection connection, JobType type, Object args, InsertOptions options) { + trace.add(name + ":insert_begin"); + } + + @Override + public List insertMany( + Connection connection, + List> submissions, + Callable> next) + throws Exception { + trace.add(name + ":insert_before"); + var results = next.call(); + trace.add(name + ":insert_after"); + return results; + } + + @Override + public void work(WorkContext context, WorkContext.Step next) throws Exception { + trace.add(name + ":work_before"); + next.run(); + trace.add(name + ":work_after"); + } + + @Override + public void workBegin(WorkContext context) { + trace.add(name + ":work_begin"); + } + + @Override + public Throwable workEnd(WorkContext context, Throwable failure) { + trace.add(name + ":work_end"); + return failure; + } + } +} diff --git a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java index e7206c185..75a7f23ad 100644 --- a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java +++ b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java @@ -66,8 +66,8 @@ void rescueCountAndOutput() throws Exception { var maintained = new CompletableFuture(); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public boolean clean(Client client, Instant now) { // Maintenance calls clean only after rescue has committed. diff --git a/java/river/src/test/java/com/riverqueue/RescueTest.java b/java/river/src/test/java/com/riverqueue/RescueTest.java index dc424600b..f85a5ecfb 100644 --- a/java/river/src/test/java/com/riverqueue/RescueTest.java +++ b/java/river/src/test/java/com/riverqueue/RescueTest.java @@ -35,8 +35,8 @@ void shortRescuedRetriesWakeFetchingBeforeMaintenance(String mode) throws Except var failClaim = new AtomicBoolean(mode.equals("peer_claim_retry")); var client = new Client(failingNotifications(database, maintained, failNotification)) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public boolean clean(Client client, Instant now) { maintained.complete(null); @@ -64,8 +64,8 @@ public boolean clean(Client client, Instant now) { var fetched = new CompletableFuture(); var peerClient = new Client(database) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public List claim( Connection connection, diff --git a/java/river/src/test/java/com/riverqueue/RetryTest.java b/java/river/src/test/java/com/riverqueue/RetryTest.java index ef5a0418b..52385d523 100644 --- a/java/river/src/test/java/com/riverqueue/RetryTest.java +++ b/java/river/src/test/java/com/riverqueue/RetryTest.java @@ -75,8 +75,8 @@ void retryWakesWorkersAfterCommit(String mode) throws Exception { var idle = new CompletableFuture(); var client = new Client(database) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterAttempt( Connection connection, Client.Driver driver, Job original) { diff --git a/java/river/src/test/java/com/riverqueue/SchedulingTest.java b/java/river/src/test/java/com/riverqueue/SchedulingTest.java index 85a6be311..e6f273eb4 100644 --- a/java/river/src/test/java/com/riverqueue/SchedulingTest.java +++ b/java/river/src/test/java/com/riverqueue/SchedulingTest.java @@ -37,8 +37,8 @@ void failedPeriodicJobDoesNotBlockMaintenanceOrOtherJobs(String failurePoint) th var worked = new LinkedBlockingQueue(); var client = new Client(databases.open(directory.resolve("river.db"))) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterInsert( Connection connection, Client.Driver driver, Job job) { @@ -95,8 +95,8 @@ void failedScheduleDoesNotInsertTheOccurrenceTwice() throws Exception { var errors = new LinkedBlockingQueue(); var client = new Client(databases.open(directory.resolve("river.db"))) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void maintain(Client client, Instant now) { if (calls.get() >= 3) maintained.complete(null); @@ -138,8 +138,8 @@ void periodicSchedulesKeepTheirPlannedCadence() throws Exception { var inserted = new AtomicReference>(); var client = new Client(databases.open(directory.resolve("river.db"))) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterInsert( Connection connection, Client.Driver driver, Job job) { diff --git a/java/river/src/test/java/com/riverqueue/VirtualThreadTest.java b/java/river/src/test/java/com/riverqueue/VirtualThreadTest.java index b0a32c231..081488b3f 100644 --- a/java/river/src/test/java/com/riverqueue/VirtualThreadTest.java +++ b/java/river/src/test/java/com/riverqueue/VirtualThreadTest.java @@ -68,8 +68,8 @@ public static void main(String[] args) throws Exception { new Migrator(database).migrate(); var client = new Client(database) - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public List claim( Connection connection, @@ -84,10 +84,10 @@ public List claim( return next.run(connection); } }) - .withExtension( - new Extension() { + .withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { if (!args[0].equals("leadership")) return; entered.countDown(); try { diff --git a/java/river/src/test/java/com/riverqueue/WorkersTest.java b/java/river/src/test/java/com/riverqueue/WorkersTest.java index e4fe2d80b..fc3a7c56d 100644 --- a/java/river/src/test/java/com/riverqueue/WorkersTest.java +++ b/java/river/src/test/java/com/riverqueue/WorkersTest.java @@ -180,8 +180,8 @@ void completionRetriesSurviveAnInterruptedHandler() throws Exception { var attempts = new AtomicInteger(); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterAttempt( Connection connection, @@ -219,8 +219,8 @@ void completionRetriesSurviveInvalidJobs(String invalid) throws Exception { var attempts = new AtomicInteger(); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterAttempt( Connection connection, @@ -235,7 +235,7 @@ public void afterAttempt( public java.util.List claim( Connection connection, Client.Driver driver, - Plugin.Claim claim, + Pilot.Claim claim, Client.Transaction> next) throws Exception { var rows = next.run(connection); @@ -294,8 +294,8 @@ void completedHandlerIsNotCancelledWhileAcknowledging() throws Exception { var context = new CompletableFuture>(); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void afterAttempt( Connection connection, @@ -329,8 +329,8 @@ void jobTimeoutDoesNotDependOnTheDispatcher() throws Exception { var release = new CountDownLatch(1); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public void producer( Connection connection, Client.Driver driver, Producer producer) @@ -437,11 +437,11 @@ void rescueCannotPreemptAConfiguredTimeout() { @ParameterizedTest @ValueSource(ints = {-1, 0, 120}) void rescueDefaultAccountsForTheJobTimeout(int timeoutMinutes) throws Exception { - var rescue = new CompletableFuture(); + var rescue = new CompletableFuture(); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public boolean rescue(Client client, Rescue request) { rescue.complete(request); @@ -542,10 +542,10 @@ void goResignationWakesLeadershipBeforeNextPoll() throws Exception { } }); var observer = - client.withExtension( - new Extension() { + client.withHooks( + new HookPeriodicJobsStart() { @Override - public void periodicStarted() { + public void start() { elected.countDown(); } }); @@ -788,8 +788,8 @@ void stoppingDuringAClaimPreservesCancellation(boolean cancel) throws Exception var commit = new CountDownLatch(1); var client = client() - .withPlugin( - new Plugin() { + .withPilot( + new Pilot() { @Override public java.util.List claim( Connection connection,