Changed the InboxCleanupService and OutboxDeliveryService registrations and configuration order to work properly

phatboyg · phatboyg · commit 89468e3e74fb · 2025-09-06T08:54:37.000-05:00
diff --git a/src/MassTransit/Configuration/DependencyInjection/InboxCleanupServiceOptions.cs b/src/MassTransit/Configuration/DependencyInjection/InboxCleanupServiceOptions.cs
@@ -25,4 +25,11 @@ public class InboxCleanupServiceOptions
         /// </summary>
         public TimeSpan QueryDelay { get; set; } = TimeSpan.FromSeconds(10);
     }
+
+
+    public class InboxCleanupServiceOptions<T> :
+        InboxCleanupServiceOptions
+        where T : class
+    {
+    }
 }
diff --git a/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/Configuration/Configuration/EntityFrameworkBusOutboxConfigurator.cs b/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/Configuration/Configuration/EntityFrameworkBusOutboxConfigurator.cs
@@ -15,11 +15,14 @@ public class EntityFrameworkBusOutboxConfigurator<TDbContext> :
     {
         readonly IBusRegistrationConfigurator _configurator;
         readonly EntityFrameworkOutboxConfigurator<TDbContext> _outboxConfigurator;
+        bool _registerOutboxDeliveryService;
 
         public EntityFrameworkBusOutboxConfigurator(IBusRegistrationConfigurator configurator, EntityFrameworkOutboxConfigurator<TDbContext> outboxConfigurator)
         {
             _outboxConfigurator = outboxConfigurator;
             _configurator = configurator;
+
+            _registerOutboxDeliveryService = true;
         }
 
         /// <summary>
@@ -34,26 +37,29 @@ public EntityFrameworkBusOutboxConfigurator(IBusRegistrationConfigurator configu
 
         public void DisableDeliveryService()
         {
-            _configurator.RemoveHostedService<BusOutboxDeliveryService<TDbContext>>();
+            _registerOutboxDeliveryService = false;
         }
 
         public virtual void Configure(Action<IEntityFrameworkBusOutboxConfigurator>? configure)
         {
-            _configurator.AddHostedService<BusOutboxDeliveryService<TDbContext>>();
+            configure?.Invoke(this);
+
             _configurator.ReplaceScoped<IScopedBusContextProvider<IBus>, EntityFrameworkScopedBusContextProvider<IBus, TDbContext>>();
             _configurator.AddSingleton<IBusOutboxNotification, BusOutboxNotification>();
 
-            _configurator.AddOptions<OutboxDeliveryServiceOptions>()
-                .Configure(options =>
-                {
-                    options.QueryDelay = _outboxConfigurator.QueryDelay;
-                    options.QueryMessageLimit = _outboxConfigurator.QueryMessageLimit;
-                    options.QueryTimeout = _outboxConfigurator.QueryTimeout;
-                    options.MessageDeliveryLimit = MessageDeliveryLimit;
-                    options.MessageDeliveryTimeout = MessageDeliveryTimeout;
-                });
-
-            configure?.Invoke(this);
+            if (_registerOutboxDeliveryService)
+            {
+                _configurator.AddHostedService<BusOutboxDeliveryService<TDbContext>>();
+                _configurator.AddOptions<OutboxDeliveryServiceOptions>()
+                    .Configure(options =>
+                    {
+                        options.QueryDelay = _outboxConfigurator.QueryDelay;
+                        options.QueryMessageLimit = _outboxConfigurator.QueryMessageLimit;
+                        options.QueryTimeout = _outboxConfigurator.QueryTimeout;
+                        options.MessageDeliveryLimit = MessageDeliveryLimit;
+                        options.MessageDeliveryTimeout = MessageDeliveryTimeout;
+                    });
+            }
         }
     }
 }
diff --git a/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/Configuration/Configuration/EntityFrameworkOutboxConfigurator.cs b/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/Configuration/Configuration/EntityFrameworkOutboxConfigurator.cs
@@ -17,13 +17,15 @@ public class EntityFrameworkOutboxConfigurator<TDbContext> :
         readonly IBusRegistrationConfigurator _configurator;
         IsolationLevel _isolationLevel;
         ILockStatementProvider _lockStatementProvider;
+        bool _registerInboxCleanupService;
 
         public EntityFrameworkOutboxConfigurator(IBusRegistrationConfigurator configurator)
         {
             _configurator = configurator;
 
             _lockStatementProvider = new SqlServerLockStatementProvider();
             _isolationLevel = IsolationLevel.RepeatableRead;
+            _registerInboxCleanupService = true;
         }
 
         public TimeSpan DuplicateDetectionWindow { get; set; } = TimeSpan.FromMinutes(30);
@@ -46,7 +48,7 @@ public ILockStatementProvider LockStatementProvider
 
         public void DisableInboxCleanupService()
         {
-            _configurator.RemoveHostedService<InboxCleanupService<TDbContext>>();
+            _registerInboxCleanupService = false;
         }
 
         public virtual void UseBusOutbox(Action<IEntityFrameworkBusOutboxConfigurator>? configure = null)
@@ -58,23 +60,26 @@ public virtual void UseBusOutbox(Action<IEntityFrameworkBusOutboxConfigurator>?
 
         public virtual void Configure(Action<IEntityFrameworkOutboxConfigurator>? configure)
         {
+            configure?.Invoke(this);
+
             _configurator.TryAddScoped<IOutboxContextFactory<TDbContext>, EntityFrameworkOutboxContextFactory<TDbContext>>();
-            _configurator.AddOptions<EntityFrameworkOutboxOptions>().Configure(options =>
+            _configurator.AddOptions<EntityFrameworkOutboxOptions<TDbContext>>().Configure(options =>
             {
                 options.IsolationLevel = _isolationLevel;
                 options.LockStatementProvider = _lockStatementProvider;
             });
 
-            _configurator.AddHostedService<InboxCleanupService<TDbContext>>();
-            _configurator.AddOptions<InboxCleanupServiceOptions>().Configure(options =>
+            if (_registerInboxCleanupService)
             {
-                options.DuplicateDetectionWindow = DuplicateDetectionWindow;
-                options.QueryMessageLimit = QueryMessageLimit;
-                options.QueryDelay = QueryDelay;
-                options.QueryTimeout = QueryTimeout;
-            });
-
-            configure?.Invoke(this);
+                _configurator.AddHostedService<InboxCleanupService<TDbContext>>();
+                _configurator.AddOptions<InboxCleanupServiceOptions<TDbContext>>().Configure(options =>
+                {
+                    options.DuplicateDetectionWindow = DuplicateDetectionWindow;
+                    options.QueryMessageLimit = QueryMessageLimit;
+                    options.QueryDelay = QueryDelay;
+                    options.QueryTimeout = QueryTimeout;
+                });
+            }
         }
     }
 }
diff --git a/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/BusOutboxDeliveryService.cs b/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/BusOutboxDeliveryService.cs
@@ -38,7 +38,7 @@ public class BusOutboxDeliveryService<TDbContext> :
         string _getOutboxIdStatement;
 
         public BusOutboxDeliveryService(IBusControl busControl, IOptions<OutboxDeliveryServiceOptions> options,
-            IOptions<EntityFrameworkOutboxOptions> outboxOptions, IBusOutboxNotification notification,
+            IOptions<EntityFrameworkOutboxOptions<TDbContext>> outboxOptions, IBusOutboxNotification notification,
             ILogger<BusOutboxDeliveryService<TDbContext>> logger, IServiceProvider provider)
         {
             _busControl = busControl;
diff --git a/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/EntityFrameworkOutboxContextFactory.cs b/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/EntityFrameworkOutboxContextFactory.cs
@@ -22,7 +22,7 @@ public class EntityFrameworkOutboxContextFactory<TDbContext> :
         readonly IServiceProvider _provider;
         string _lockStatement;
 
-        public EntityFrameworkOutboxContextFactory(TDbContext dbContext, IServiceProvider provider, IOptions<EntityFrameworkOutboxOptions> options)
+        public EntityFrameworkOutboxContextFactory(TDbContext dbContext, IServiceProvider provider, IOptions<EntityFrameworkOutboxOptions<TDbContext>> options)
         {
             _dbContext = dbContext;
             _provider = provider;
diff --git a/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/InboxCleanupService.cs b/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/InboxCleanupService.cs
@@ -22,11 +22,11 @@ public class InboxCleanupService<TDbContext> :
         where TDbContext : DbContext
     {
         readonly ILogger<InboxCleanupService<TDbContext>> _logger;
-        readonly InboxCleanupServiceOptions _options;
+        readonly InboxCleanupServiceOptions<TDbContext> _options;
         readonly IServiceProvider _provider;
         readonly IRetryPolicy _retryPolicy;
 
-        public InboxCleanupService(IOptions<InboxCleanupServiceOptions> options, ILogger<InboxCleanupService<TDbContext>> logger,
+        public InboxCleanupService(IOptions<InboxCleanupServiceOptions<TDbContext>> options, ILogger<InboxCleanupService<TDbContext>> logger,
             IServiceProvider provider)
         {
             _options = options.Value;
diff --git a/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkOutboxOptions.cs b/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkOutboxOptions.cs
@@ -2,9 +2,11 @@ namespace MassTransit
 {
     using System.Data;
     using EntityFrameworkCoreIntegration;
+    using Microsoft.EntityFrameworkCore;
 
 
-    public class EntityFrameworkOutboxOptions
+    public class EntityFrameworkOutboxOptions<TDbContext>
+        where TDbContext : DbContext
     {
         public IsolationLevel IsolationLevel { get; set; } = IsolationLevel.RepeatableRead;
         public ILockStatementProvider LockStatementProvider { get; set; } = new SqlServerLockStatementProvider();
diff --git a/src/Persistence/MassTransit.MongoDbIntegration/Configuration/Configuration/MongoDbBusOutboxConfigurator.cs b/src/Persistence/MassTransit.MongoDbIntegration/Configuration/Configuration/MongoDbBusOutboxConfigurator.cs
@@ -13,11 +13,14 @@ public class MongoDbBusOutboxConfigurator :
     {
         readonly IBusRegistrationConfigurator _configurator;
         readonly MongoDbOutboxConfigurator _outboxConfigurator;
+        bool _registerOutboxDeliveryService;
 
         public MongoDbBusOutboxConfigurator(IBusRegistrationConfigurator configurator, MongoDbOutboxConfigurator outboxConfigurator)
         {
             _outboxConfigurator = outboxConfigurator;
             _configurator = configurator;
+
+            _registerOutboxDeliveryService = true;
         }
 
         /// <summary>
@@ -32,26 +35,29 @@ public MongoDbBusOutboxConfigurator(IBusRegistrationConfigurator configurator, M
 
         public void DisableDeliveryService()
         {
-            _configurator.RemoveHostedService<BusOutboxDeliveryService>();
+            _registerOutboxDeliveryService = false;
         }
 
         public virtual void Configure(Action<IMongoDbBusOutboxConfigurator>? configure)
         {
-            _configurator.AddHostedService<BusOutboxDeliveryService>();
+            configure?.Invoke(this);
+
             _configurator.ReplaceScoped<IScopedBusContextProvider<IBus>, MongoDbScopedBusContextProvider<IBus>>();
             _configurator.AddSingleton<IBusOutboxNotification, BusOutboxNotification>();
 
-            _configurator.AddOptions<OutboxDeliveryServiceOptions>()
-                .Configure(options =>
-                {
-                    options.QueryDelay = _outboxConfigurator.QueryDelay;
-                    options.QueryMessageLimit = _outboxConfigurator.QueryMessageLimit;
-                    options.QueryTimeout = _outboxConfigurator.QueryTimeout;
-                    options.MessageDeliveryLimit = MessageDeliveryLimit;
-                    options.MessageDeliveryTimeout = MessageDeliveryTimeout;
-                });
-
-            configure?.Invoke(this);
+            if (_registerOutboxDeliveryService)
+            {
+                _configurator.AddHostedService<BusOutboxDeliveryService>();
+                _configurator.AddOptions<OutboxDeliveryServiceOptions>()
+                    .Configure(options =>
+                    {
+                        options.QueryDelay = _outboxConfigurator.QueryDelay;
+                        options.QueryMessageLimit = _outboxConfigurator.QueryMessageLimit;
+                        options.QueryTimeout = _outboxConfigurator.QueryTimeout;
+                        options.MessageDeliveryLimit = MessageDeliveryLimit;
+                        options.MessageDeliveryTimeout = MessageDeliveryTimeout;
+                    });
+            }
         }
     }
 }
diff --git a/src/Persistence/MassTransit.MongoDbIntegration/Configuration/Configuration/MongoDbOutboxConfigurator.cs b/src/Persistence/MassTransit.MongoDbIntegration/Configuration/Configuration/MongoDbOutboxConfigurator.cs
@@ -16,10 +16,13 @@ public class MongoDbOutboxConfigurator :
         IMongoDbOutboxConfigurator
     {
         readonly IBusRegistrationConfigurator _configurator;
+        bool _registerInboxCleanupService;
 
         public MongoDbOutboxConfigurator(IBusRegistrationConfigurator configurator)
         {
             _configurator = configurator;
+
+            _registerInboxCleanupService = true;
         }
 
         public TimeSpan DuplicateDetectionWindow { get; set; } = TimeSpan.FromMinutes(30);
@@ -32,7 +35,7 @@ public MongoDbOutboxConfigurator(IBusRegistrationConfigurator configurator)
 
         public void DisableInboxCleanupService()
         {
-            _configurator.RemoveHostedService<InboxCleanupService>();
+            _registerInboxCleanupService = false;
         }
 
         public virtual void UseBusOutbox(Action<IMongoDbBusOutboxConfigurator>? configure = null)
@@ -51,14 +54,17 @@ public virtual void Configure(Action<IMongoDbOutboxConfigurator>? configure)
 
             _configurator.TryAddScoped<IOutboxContextFactory<MongoDbContext>, MongoDbOutboxContextFactory>();
 
-            _configurator.AddHostedService<InboxCleanupService>();
-            _configurator.AddOptions<InboxCleanupServiceOptions>().Configure(options =>
+            if (_registerInboxCleanupService)
             {
-                options.DuplicateDetectionWindow = DuplicateDetectionWindow;
-                options.QueryMessageLimit = QueryMessageLimit;
-                options.QueryDelay = QueryDelay;
-                options.QueryTimeout = QueryTimeout;
-            });
+                _configurator.AddHostedService<InboxCleanupService>();
+                _configurator.AddOptions<InboxCleanupServiceOptions>().Configure(options =>
+                {
+                    options.DuplicateDetectionWindow = DuplicateDetectionWindow;
+                    options.QueryMessageLimit = QueryMessageLimit;
+                    options.QueryDelay = QueryDelay;
+                    options.QueryTimeout = QueryTimeout;
+                });
+            }
 
             RegisterClassMaps();
 

Original file line number	Diff line number	Diff line change
`@@ -25,4 +25,11 @@ public class InboxCleanupServiceOptions`
`25`	`25`	`/// </summary>`
`26`	`26`	`public TimeSpan QueryDelay { get; set; } = TimeSpan.FromSeconds(10);`
`27`	`27`	`}`
	`28`	`+`
	`29`	`+`
	`30`	`+ public class InboxCleanupServiceOptions<T> :`
	`31`	`+ InboxCleanupServiceOptions`
	`32`	`+ where T : class`
	`33`	`+ {`
	`34`	`+ }`
`28`	`35`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,13 +17,15 @@ public class EntityFrameworkOutboxConfigurator<TDbContext> :`
`17`	`17`	`readonly IBusRegistrationConfigurator _configurator;`
`18`	`18`	`IsolationLevel _isolationLevel;`
`19`	`19`	`ILockStatementProvider _lockStatementProvider;`
	`20`	`+ bool _registerInboxCleanupService;`
`20`	`21`
`21`	`22`	`public EntityFrameworkOutboxConfigurator(IBusRegistrationConfigurator configurator)`
`22`	`23`	`{`
`23`	`24`	`_configurator = configurator;`
`24`	`25`
`25`	`26`	`_lockStatementProvider = new SqlServerLockStatementProvider();`
`26`	`27`	`_isolationLevel = IsolationLevel.RepeatableRead;`
	`28`	`+ _registerInboxCleanupService = true;`
`27`	`29`	`}`
`28`	`30`
`29`	`31`	`public TimeSpan DuplicateDetectionWindow { get; set; } = TimeSpan.FromMinutes(30);`
`@@ -46,7 +48,7 @@ public ILockStatementProvider LockStatementProvider`
`46`	`48`
`47`	`49`	`public void DisableInboxCleanupService()`
`48`	`50`	`{`
`49`		`- _configurator.RemoveHostedService<InboxCleanupService<TDbContext>>();`
	`51`	`+ _registerInboxCleanupService = false;`
`50`	`52`	`}`
`51`	`53`
`52`	`54`	`public virtual void UseBusOutbox(Action<IEntityFrameworkBusOutboxConfigurator>? configure = null)`
`@@ -58,23 +60,26 @@ public virtual void UseBusOutbox(Action<IEntityFrameworkBusOutboxConfigurator>?`
`58`	`60`
`59`	`61`	`public virtual void Configure(Action<IEntityFrameworkOutboxConfigurator>? configure)`
`60`	`62`	`{`
	`63`	`+ configure?.Invoke(this);`
	`64`	`+`
`61`	`65`	`_configurator.TryAddScoped<IOutboxContextFactory<TDbContext>, EntityFrameworkOutboxContextFactory<TDbContext>>();`
`62`		`- _configurator.AddOptions<EntityFrameworkOutboxOptions>().Configure(options =>`
	`66`	`+ _configurator.AddOptions<EntityFrameworkOutboxOptions<TDbContext>>().Configure(options =>`
`63`	`67`	`{`
`64`	`68`	`options.IsolationLevel = _isolationLevel;`
`65`	`69`	`options.LockStatementProvider = _lockStatementProvider;`
`66`	`70`	`});`
`67`	`71`
`68`		`- _configurator.AddHostedService<InboxCleanupService<TDbContext>>();`
`69`		`- _configurator.AddOptions<InboxCleanupServiceOptions>().Configure(options =>`
	`72`	`+ if (_registerInboxCleanupService)`
`70`	`73`	`{`
`71`		`- options.DuplicateDetectionWindow = DuplicateDetectionWindow;`
`72`		`- options.QueryMessageLimit = QueryMessageLimit;`
`73`		`- options.QueryDelay = QueryDelay;`
`74`		`- options.QueryTimeout = QueryTimeout;`
`75`		`- });`
`76`		`-`
`77`		`- configure?.Invoke(this);`
	`74`	`+ _configurator.AddHostedService<InboxCleanupService<TDbContext>>();`
	`75`	`+ _configurator.AddOptions<InboxCleanupServiceOptions<TDbContext>>().Configure(options =>`
	`76`	`+ {`
	`77`	`+ options.DuplicateDetectionWindow = DuplicateDetectionWindow;`
	`78`	`+ options.QueryMessageLimit = QueryMessageLimit;`
	`79`	`+ options.QueryDelay = QueryDelay;`
	`80`	`+ options.QueryTimeout = QueryTimeout;`
	`81`	`+ });`
	`82`	`+ }`
`78`	`83`	`}`
`79`	`84`	`}`
`80`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ public class BusOutboxDeliveryService<TDbContext> :`
`38`	`38`	`string _getOutboxIdStatement;`
`39`	`39`
`40`	`40`	`public BusOutboxDeliveryService(IBusControl busControl, IOptions<OutboxDeliveryServiceOptions> options,`
`41`		`- IOptions<EntityFrameworkOutboxOptions> outboxOptions, IBusOutboxNotification notification,`
	`41`	`+ IOptions<EntityFrameworkOutboxOptions<TDbContext>> outboxOptions, IBusOutboxNotification notification,`
`42`	`42`	`ILogger<BusOutboxDeliveryService<TDbContext>> logger, IServiceProvider provider)`
`43`	`43`	`{`
`44`	`44`	`_busControl = busControl;`
Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,7 @@ public class EntityFrameworkOutboxContextFactory<TDbContext> :`
`22`	`22`	`readonly IServiceProvider _provider;`
`23`	`23`	`string _lockStatement;`
`24`	`24`
`25`		`- public EntityFrameworkOutboxContextFactory(TDbContext dbContext, IServiceProvider provider, IOptions<EntityFrameworkOutboxOptions> options)`
	`25`	`+ public EntityFrameworkOutboxContextFactory(TDbContext dbContext, IServiceProvider provider, IOptions<EntityFrameworkOutboxOptions<TDbContext>> options)`
`26`	`26`	`{`
`27`	`27`	`_dbContext = dbContext;`
`28`	`28`	`_provider = provider;`
Original file line number	Diff line number	Diff line change
`@@ -22,11 +22,11 @@ public class InboxCleanupService<TDbContext> :`
`22`	`22`	`where TDbContext : DbContext`
`23`	`23`	`{`
`24`	`24`	`readonly ILogger<InboxCleanupService<TDbContext>> _logger;`
`25`		`- readonly InboxCleanupServiceOptions _options;`
	`25`	`+ readonly InboxCleanupServiceOptions<TDbContext> _options;`
`26`	`26`	`readonly IServiceProvider _provider;`
`27`	`27`	`readonly IRetryPolicy _retryPolicy;`
`28`	`28`
`29`		`- public InboxCleanupService(IOptions<InboxCleanupServiceOptions> options, ILogger<InboxCleanupService<TDbContext>> logger,`
	`29`	`+ public InboxCleanupService(IOptions<InboxCleanupServiceOptions<TDbContext>> options, ILogger<InboxCleanupService<TDbContext>> logger,`
`30`	`30`	`IServiceProvider provider)`
`31`	`31`	`{`
`32`	`32`	`_options = options.Value;`
Original file line number	Diff line number	Diff line change
`@@ -2,9 +2,11 @@ namespace MassTransit`
`2`	`2`	`{`
`3`	`3`	`using System.Data;`
`4`	`4`	`using EntityFrameworkCoreIntegration;`
	`5`	`+ using Microsoft.EntityFrameworkCore;`
`5`	`6`
`6`	`7`
`7`		`- public class EntityFrameworkOutboxOptions`
	`8`	`+ public class EntityFrameworkOutboxOptions<TDbContext>`
	`9`	`+ where TDbContext : DbContext`
`8`	`10`	`{`
`9`	`11`	`public IsolationLevel IsolationLevel { get; set; } = IsolationLevel.RepeatableRead;`
`10`	`12`	`public ILockStatementProvider LockStatementProvider { get; set; } = new SqlServerLockStatementProvider();`