diff --git a/CHANGELOG.md b/CHANGELOG.md index 3792e801..b72b93de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,8 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Both hooks and middleware should now be registered under a general `Config.Plugins` instead, a change that simpifies things somewhat, and better handles cases where a hook or middleware might be _both_ a hook and a middleware as opposed to only one or the other. `Config.Hooks` and `Config.Middleware` are still available for use, but considered deprecated in favor of the more general `Config.Plugins`. [PR #1284](https://github.com/riverqueue/river/pull/1284). -- Hooks passed to `Config.Hooks` that also implement middleware now have their middleware functionality activate automatically. The converse is also true for middleware sent to `Config.Middleware` that implement hooks. [PR #1284](https://github.com/riverqueue/river/pull/1284). +- Added `Config.Plugins` for extensions that should be installed as both hooks and middleware. `Config.Hooks` and `Config.Middleware` remain available for hook-only and middleware-only registration. [PR #1284](https://github.com/riverqueue/river/pull/1284). ### Fixed diff --git a/client.go b/client.go index 96d7b3eb..422eb4d7 100644 --- a/client.go +++ b/client.go @@ -242,10 +242,9 @@ type Config struct { // // Jobs may have their own specific hooks by implementing JobArgsWithHooks. // - // If a type in Hooks also implements rivertype.Middleware, it will be - // installed as middleware too. - // - // Deprecated: Use Plugins instead. + // Entries in Hooks are installed only as hooks, even if they also implement + // rivertype.Middleware. Use Plugins for an extension that should act as + // both a hook and middleware. Hooks []rivertype.Hook // Logger is the structured logger to use for logging purposes. If none is @@ -280,10 +279,9 @@ type Config struct { // them will not run. When a job is worked, the work middleware runs and the // insertion middlewares on either side of it are skipped. // - // If a type in Middleware also implements rivertype.Hook, it will be - // installed as a hook too. - // - // Deprecated: Use Plugins instead. + // Entries in Middleware are installed only as middleware, even if they also + // implement rivertype.Hook. Use Plugins for an extension that should act as + // both middleware and a hook. Middleware []rivertype.Middleware // Plugins contains extensions installed globally as hooks, middleware, or @@ -296,8 +294,8 @@ type Config struct { // MiddlewareDefaults directly and define its own IsPlugin method, then // implement any operation-specific hook or middleware interfaces it needs. // - // Hooks and Middleware are still supported for backward compatibility, but - // Plugins is the preferred place to register new extensions. + // Use Hooks or Middleware when an extension should be installed only as the + // corresponding kind. Use Plugins when it should be eligible as both. Plugins []rivertype.Plugin // PeriodicJobs are a set of periodic jobs to run at the specified intervals @@ -829,9 +827,10 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client var ( middleware = pluginconfig.CombinedMiddleware(config.Middleware, config.JobInsertMiddleware, config.WorkerMiddleware) plugins = append(riverplugin.DefaultPlugins(), config.Plugins...) - allPlugins = pluginlookup.NormalizePlugins(config.Hooks, middleware, plugins) ) - pluginlookup.InitBaseServices(archetype, allPlugins) + pluginlookup.InitBaseServices(archetype, config.Hooks) + pluginlookup.InitBaseServices(archetype, middleware) + pluginlookup.InitBaseServices(archetype, plugins) client := &Client[TTx]{ clientNotifyBundle: &ClientNotifyBundle[TTx]{ @@ -841,7 +840,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client config: config, driver: driver, pluginLookupByJob: pluginlookup.NewJobPluginLookup(), - pluginLookupGlobal: pluginlookup.NewPluginLookup(allPlugins), + pluginLookupGlobal: pluginlookup.NewPluginLookupFromConfig(config.Hooks, middleware, plugins), producersByQueueName: make(map[string]*producer), testSignals: clientTestSignals{}, workCancel: func(cause error) {}, // replaced on start, but here in case StopAndCancel is called before start up diff --git a/client_test.go b/client_test.go index 478ff6eb..0274f5ef 100644 --- a/client_test.go +++ b/client_test.go @@ -8322,7 +8322,7 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { } } - insertAndRequireCounts := func(t *testing.T, bundle *testBundle, plugin *hookMiddlewarePlugin, expectedCount int) { + insertAndRequireCounts := func(t *testing.T, bundle *testBundle, plugin *hookMiddlewarePlugin, expectedInsertBeginCount, expectedInsertManyCount int) { t.Helper() client := newTestClient(t, bundle.dbPool, bundle.config) @@ -8330,8 +8330,8 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { _, err := client.Insert(ctx, noOpArgs{}, nil) require.NoError(t, err) - require.Equal(t, expectedCount, plugin.insertBeginCount) - require.Equal(t, expectedCount, plugin.insertManyCount) + require.Equal(t, expectedInsertBeginCount, plugin.insertBeginCount) + require.Equal(t, expectedInsertManyCount, plugin.insertManyCount) } t.Run("DuplicatePointerAcrossHooksMiddlewareAndPluginsRunsMultipleTimes", func(t *testing.T) { @@ -8343,7 +8343,7 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { bundle.config.Middleware = []rivertype.Middleware{plugin} bundle.config.Plugins = []rivertype.Plugin{plugin} - insertAndRequireCounts(t, bundle, plugin, 3) + insertAndRequireCounts(t, bundle, plugin, 2, 2) }) t.Run("DuplicatePointerWithinHooksRunsMultipleTimes", func(t *testing.T) { @@ -8353,20 +8353,20 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { plugin := &hookMiddlewarePlugin{} bundle.config.Hooks = []rivertype.Hook{plugin, plugin} - insertAndRequireCounts(t, bundle, plugin, 2) + insertAndRequireCounts(t, bundle, plugin, 2, 0) }) - t.Run("HookAlsoRegisteredAsMiddleware", func(t *testing.T) { + t.Run("HookNotRegisteredAsMiddleware", func(t *testing.T) { t.Parallel() bundle := setup(t) plugin := &hookMiddlewarePlugin{} bundle.config.Hooks = []rivertype.Hook{plugin} - insertAndRequireCounts(t, bundle, plugin, 1) + insertAndRequireCounts(t, bundle, plugin, 1, 0) }) - t.Run("LegacyHybridRegisteredInHooksAndMiddlewareRunsMultipleTimes", func(t *testing.T) { + t.Run("LegacyHybridRegisteredInHooksAndMiddlewareRunsOncePerKind", func(t *testing.T) { t.Parallel() bundle := setup(t) @@ -8380,11 +8380,11 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { require.NoError(t, err) require.NotEmpty(t, plugin.Name) - require.Equal(t, 2, plugin.insertBeginCount) - require.Equal(t, 2, plugin.insertManyCount) + require.Equal(t, 1, plugin.insertBeginCount) + require.Equal(t, 1, plugin.insertManyCount) }) - t.Run("LegacyHybridRegisteredInHooksRunsBothAndIsInitialized", func(t *testing.T) { + t.Run("LegacyHybridRegisteredInHooksRunsOnlyAsHookAndIsInitialized", func(t *testing.T) { t.Parallel() bundle := setup(t) @@ -8398,10 +8398,10 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { require.NotEmpty(t, plugin.Name) require.Equal(t, 1, plugin.insertBeginCount) - require.Equal(t, 1, plugin.insertManyCount) + require.Zero(t, plugin.insertManyCount) }) - t.Run("LegacyHybridRegisteredInMiddlewareRunsBothAndIsInitialized", func(t *testing.T) { + t.Run("LegacyHybridRegisteredInMiddlewareRunsOnlyAsMiddlewareAndIsInitialized", func(t *testing.T) { t.Parallel() bundle := setup(t) @@ -8414,18 +8414,18 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { require.NoError(t, err) require.NotEmpty(t, plugin.Name) - require.Equal(t, 1, plugin.insertBeginCount) + require.Zero(t, plugin.insertBeginCount) require.Equal(t, 1, plugin.insertManyCount) }) - t.Run("MiddlewareAlsoRegisteredAsHook", func(t *testing.T) { + t.Run("MiddlewareNotRegisteredAsHook", func(t *testing.T) { t.Parallel() bundle := setup(t) plugin := &hookMiddlewarePlugin{} bundle.config.Middleware = []rivertype.Middleware{plugin} - insertAndRequireCounts(t, bundle, plugin, 1) + insertAndRequireCounts(t, bundle, plugin, 0, 1) }) t.Run("PluginRegisteredAsHookAndMiddleware", func(t *testing.T) { @@ -8435,7 +8435,7 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { plugin := &hookMiddlewarePlugin{} bundle.config.Plugins = []rivertype.Plugin{plugin} - insertAndRequireCounts(t, bundle, plugin, 1) + insertAndRequireCounts(t, bundle, plugin, 1, 1) }) t.Run("PluginRegisteredWithEmbeddedDefaults", func(t *testing.T) { @@ -8471,8 +8471,8 @@ func Test_NewClient_PluginsAndHybrids(t *testing.T) { _, err := client.Insert(ctx, noOpArgs{}, nil) require.NoError(t, err) - require.Equal(t, 2, counts.insertBeginCount) - require.Equal(t, 2, counts.insertManyCount) + require.Equal(t, 1, counts.insertBeginCount) + require.Equal(t, 1, counts.insertManyCount) }) } diff --git a/internal/pluginlookup/plugin_lookup.go b/internal/pluginlookup/plugin_lookup.go index 36d83286..9829916c 100644 --- a/internal/pluginlookup/plugin_lookup.go +++ b/internal/pluginlookup/plugin_lookup.go @@ -23,33 +23,15 @@ const ( PluginKindMiddlewareWorker PluginKind = "middleware_worker" ) -// InitBaseServices initializes base services embedded in normalized plugins, -// hooks, and middleware. -func InitBaseServices(archetype *baseservice.Archetype, plugins []any) { +// InitBaseServices initializes base services embedded in plugins. +func InitBaseServices[T any](archetype *baseservice.Archetype, plugins []T) { for _, plugin := range plugins { - if withBaseService, ok := plugin.(baseservice.WithBaseService); ok { + if withBaseService, ok := any(plugin).(baseservice.WithBaseService); ok { baseservice.Init(archetype, withBaseService) } } } -// NormalizePlugins combines configured hooks, middleware, and plugins into a -// single plugin slice while preserving legacy hooks and middleware that don't -// yet opt into Plugin. -// -// Plugins passed explicitly are ordered before hooks and middleware, so -// Config.Plugins entries take precedence over plugins bridged from Config.Hooks -// or Config.Middleware. Relative order is preserved within each input slice, -// and repeated values are preserved wherever they occur. -func NormalizePlugins(hooks []rivertype.Hook, middlewares []rivertype.Middleware, plugins []rivertype.Plugin) []any { - normalizedPlugins := make([]any, 0, len(hooks)+len(middlewares)+len(plugins)) - normalizedPlugins = append(normalizedPlugins, toAnySlice(plugins)...) - normalizedPlugins = append(normalizedPlugins, toAnySlice(hooks)...) - normalizedPlugins = append(normalizedPlugins, toAnySlice(middlewares)...) - - return normalizedPlugins -} - // // PluginLookupInterface // @@ -64,21 +46,43 @@ type PluginLookupInterface interface { // NewPluginLookup returns a new plugin lookup interface based on the given // plugins that satisfies PluginLookupInterface. This is often pluginLookup, // but may be emptyPluginLookup as an optimization for the common case of an -// empty plugin bundle. +// empty plugin bundle. Each input is considered for every hook and middleware +// kind it implements. // // The plugins parameter is []any rather than []rivertype.Plugin because the -// normalized plugin list may contain legacy hooks and middleware that don't -// implement rivertype.Plugin. Keeping their original concrete values avoids -// compatibility wrappers that would need to forward every operation-specific -// interface, along with auxiliary interfaces like baseservice.WithBaseService. +// lookup may contain legacy hooks and middleware that don't implement +// rivertype.Plugin. Keeping their original concrete values avoids compatibility +// wrappers that would need to forward every operation-specific interface. func NewPluginLookup(plugins []any) PluginLookupInterface { - if len(plugins) < 1 { + return newPluginLookup(plugins, plugins) +} + +// NewPluginLookupFromConfig returns a plugin lookup from separately configured +// hooks, middleware, and plugins. Explicit plugins may participate as either +// hooks or middleware, while entries from the legacy Hooks and Middleware +// configuration fields participate only as the kind they were configured as. +func NewPluginLookupFromConfig(hooks []rivertype.Hook, middlewares []rivertype.Middleware, plugins []rivertype.Plugin) PluginLookupInterface { + pluginValues := toAnySlice(plugins) + + hookValues := make([]any, 0, len(plugins)+len(hooks)) + hookValues = append(hookValues, pluginValues...) + hookValues = append(hookValues, toAnySlice(hooks)...) + + middlewareValues := make([]any, 0, len(plugins)+len(middlewares)) + middlewareValues = append(middlewareValues, pluginValues...) + middlewareValues = append(middlewareValues, toAnySlice(middlewares)...) + + return newPluginLookup(hookValues, middlewareValues) +} + +func newPluginLookup(hooks, middlewares []any) PluginLookupInterface { + if len(hooks) < 1 && len(middlewares) < 1 { return &emptyPluginLookup{} } pluginsByKind := make(map[PluginKind][]any) - for _, plugin := range plugins { + for _, plugin := range hooks { if plugin == nil { continue } @@ -98,6 +102,13 @@ func NewPluginLookup(plugins []any) PluginLookupInterface { if _, ok := plugin.(rivertype.HookWorkEnd); ok { pluginsByKind[PluginKindHookWorkEnd] = append(pluginsByKind[PluginKindHookWorkEnd], plugin) } + } + + for _, plugin := range middlewares { + if plugin == nil { + continue + } + if _, ok := plugin.(rivertype.JobInsertMiddleware); ok { pluginsByKind[PluginKindMiddlewareJobInsert] = append(pluginsByKind[PluginKindMiddlewareJobInsert], plugin) } @@ -185,7 +196,7 @@ func (c *JobPluginLookup) ByJobArgs(args rivertype.JobArgs) PluginLookupInterfac hooks = argsWithHooks.Hooks() } - lookup = NewPluginLookup(toAnySlice(hooks)) + lookup = newPluginLookup(toAnySlice(hooks), nil) c.pluginLookupByKind[kind] = lookup return lookup } diff --git a/internal/pluginlookup/plugin_lookup_test.go b/internal/pluginlookup/plugin_lookup_test.go index 3d9484bb..1749927c 100644 --- a/internal/pluginlookup/plugin_lookup_test.go +++ b/internal/pluginlookup/plugin_lookup_test.go @@ -111,82 +111,28 @@ func TestJobPluginLookup(t *testing.T) { }) } -func TestNormalizePlugins(t *testing.T) { +func TestNewPluginLookupFromConfig(t *testing.T) { t.Parallel() - t.Run("DistinctZeroSizedPointersArePreserved", func(t *testing.T) { - t.Parallel() - - plugins := NormalizePlugins(nil, nil, []rivertype.Plugin{ - &testZeroSizedHookMiddlewarePlugin{}, - &testZeroSizedHookMiddlewarePlugin{}, - }) - - require.Len(t, plugins, 2) - }) - - t.Run("DuplicatesWithinAGroupArePreserved", func(t *testing.T) { - t.Parallel() - - plugin := &testHookMiddlewarePlugin{} - plugins := NormalizePlugins(nil, nil, []rivertype.Plugin{plugin, plugin}) - - require.Equal(t, []any{plugin, plugin}, plugins) - }) - - t.Run("PluginsPrecedeHooksAndMiddleware", func(t *testing.T) { - t.Parallel() - - hookPlugin := &testHookMiddlewarePlugin{} - middlewarePlugin := &testHookMiddlewarePlugin{} - plugin := &testHookMiddlewarePlugin{} - - plugins := NormalizePlugins( - []rivertype.Hook{hookPlugin}, - []rivertype.Middleware{middlewarePlugin}, - []rivertype.Plugin{plugin}, - ) - - lookup, isPluginLookup := NewPluginLookup(plugins).(*pluginLookup) - require.True(t, isPluginLookup) - - require.Equal(t, []any{ - plugin, - hookPlugin, - middlewarePlugin, - }, lookup.ByKind(PluginKindHookInsertBegin)) - require.Equal(t, []any{ - plugin, - hookPlugin, - middlewarePlugin, - }, lookup.ByKind(PluginKindMiddlewareJobInsert)) - }) - - t.Run("SamePointerAcrossGroupsIsPreserved", func(t *testing.T) { - t.Parallel() - - plugin := &testHookMiddlewarePlugin{} - plugins := NormalizePlugins( - []rivertype.Hook{plugin}, - []rivertype.Middleware{plugin}, - []rivertype.Plugin{plugin}, - ) - - require.Equal(t, []any{plugin, plugin, plugin}, plugins) - }) - - t.Run("SameZeroSizedPointerAcrossGroupsIsPreserved", func(t *testing.T) { - t.Parallel() - - plugin := &testZeroSizedHookMiddlewarePlugin{} - plugins := NormalizePlugins( - []rivertype.Hook{plugin}, - []rivertype.Middleware{plugin}, - nil, - ) - - require.Equal(t, []any{plugin, plugin}, plugins) - }) + hookPlugin := &testHookMiddlewarePlugin{} + middlewarePlugin := &testHookMiddlewarePlugin{} + plugin := &testHookMiddlewarePlugin{} + + lookup, isPluginLookup := NewPluginLookupFromConfig( + []rivertype.Hook{hookPlugin}, + []rivertype.Middleware{middlewarePlugin}, + []rivertype.Plugin{plugin}, + ).(*pluginLookup) + require.True(t, isPluginLookup) + + require.Equal(t, []any{ + plugin, + hookPlugin, + }, lookup.ByKind(PluginKindHookInsertBegin)) + require.Equal(t, []any{ + plugin, + middlewarePlugin, + }, lookup.ByKind(PluginKindMiddlewareJobInsert)) } func TestPluginLookup(t *testing.T) { @@ -298,13 +244,11 @@ func TestPluginLookup(t *testing.T) { legacyHook := &testLegacyHookInsertBegin{} legacyMiddleware := &testLegacyMiddlewareJobInsert{} - plugins := NormalizePlugins( + lookup, isPluginLookup := NewPluginLookupFromConfig( []rivertype.Hook{legacyHook}, []rivertype.Middleware{legacyMiddleware}, nil, - ) - - lookup, isPluginLookup := NewPluginLookup(plugins).(*pluginLookup) + ).(*pluginLookup) require.True(t, isPluginLookup) hookPlugins := lookup.ByKind(PluginKindHookInsertBegin) @@ -511,30 +455,6 @@ func (t *testMiddlewareWorker) Work(ctx context.Context, job *rivertype.JobRow, return doInner(ctx) } -// -// testZeroSizedHookMiddlewarePlugin -// - -var ( - _ rivertype.HookInsertBegin = &testZeroSizedHookMiddlewarePlugin{} - _ rivertype.JobInsertMiddleware = &testZeroSizedHookMiddlewarePlugin{} - _ rivertype.Plugin = &testZeroSizedHookMiddlewarePlugin{} -) - -type testZeroSizedHookMiddlewarePlugin struct{} - -func (t *testZeroSizedHookMiddlewarePlugin) InsertBegin(ctx context.Context, params *rivertype.JobInsertParams) error { - return nil -} - -func (t *testZeroSizedHookMiddlewarePlugin) InsertMany(ctx context.Context, manyParams []*rivertype.JobInsertParams, doInner func(context.Context) ([]*rivertype.JobInsertResult, error)) ([]*rivertype.JobInsertResult, error) { - return doInner(ctx) -} - -func (t *testZeroSizedHookMiddlewarePlugin) IsHook() bool { return true } -func (t *testZeroSizedHookMiddlewarePlugin) IsMiddleware() bool { return true } -func (t *testZeroSizedHookMiddlewarePlugin) IsPlugin() bool { return true } - // // testLegacyHookInsertBegin // diff --git a/rivertest/worker.go b/rivertest/worker.go index 75f30599..1616628c 100644 --- a/rivertest/worker.go +++ b/rivertest/worker.go @@ -148,11 +148,13 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job completer := jobcompleter.NewInlineCompleter(archetype, w.config.Schema, exec, w.client.Pilot(), subscribeCh) var ( + hooks = w.config.Hooks middleware = pluginconfig.CombinedMiddleware(w.config.Middleware, w.config.JobInsertMiddleware, w.config.WorkerMiddleware) //nolint:staticcheck plugins = append(riverplugin.DefaultPlugins(), w.config.Plugins...) - allPlugins = pluginlookup.NormalizePlugins(w.config.Hooks, middleware, plugins) //nolint:staticcheck ) - pluginlookup.InitBaseServices(archetype, allPlugins) + pluginlookup.InitBaseServices(archetype, hooks) + pluginlookup.InitBaseServices(archetype, middleware) + pluginlookup.InitBaseServices(archetype, plugins) updatedJobRow, err := exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{ ID: job.ID, @@ -202,7 +204,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job }, }, PluginLookupByJob: pluginlookup.NewJobPluginLookup(), - PluginLookupGlobal: pluginlookup.NewPluginLookup(allPlugins), + PluginLookupGlobal: pluginlookup.NewPluginLookupFromConfig(hooks, middleware, plugins), JobRow: job, ProducerCallbacks: struct { JobDone func(jobRow *rivertype.JobRow)