Add additional event that is triggered after each service broker receive operation

C0nquistadore · C0nquistadore · commit c861da59f94f · 2023-05-15T16:11:16.000+02:00
diff --git a/src/Dibix.Worker.Abstractions/BackgroundService.cs b/src/Dibix.Worker.Abstractions/BackgroundService.cs
@@ -11,7 +11,7 @@ public abstract class BackgroundService : HostedServiceListener, IHostedService,
         private Task? _executingTask;
         private CancellationTokenSource? _stoppingCts;
 
-        protected BackgroundService(IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(hostedServiceRegistrar, logger) { }
+        protected BackgroundService(IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(hostedServiceEvents, logger) { }
 
         protected sealed override Task StartListenerAsync(CancellationToken cancellationToken)
         {
diff --git a/src/Dibix.Worker.Abstractions/HostedServiceListener.cs b/src/Dibix.Worker.Abstractions/HostedServiceListener.cs
@@ -7,16 +7,16 @@ namespace Dibix.Worker.Abstractions
 {
     public abstract class HostedServiceListener : HostedService, IHostedService
     {
-        private readonly IHostedServiceRegistrar _hostedServiceRegistrar;
+        private readonly IHostedServiceEvents _hostedServiceEvents;
 
-        protected HostedServiceListener(IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(logger)
+        protected HostedServiceListener(IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(logger)
         {
-            _hostedServiceRegistrar = hostedServiceRegistrar;
+            _hostedServiceEvents = hostedServiceEvents;
         }
 
         protected sealed override async Task StartServiceAsync(CancellationToken cancellationToken)
         {
-            await _hostedServiceRegistrar.RegisterHostedService(GetType().FullName, cancellationToken).ConfigureAwait(false);
+            await _hostedServiceEvents.OnWorkerStarted(GetType().FullName, cancellationToken).ConfigureAwait(false);
             await StartListenerAsync(cancellationToken).ConfigureAwait(false);
         }
 
diff --git a/src/Dibix.Worker.Abstractions/IHostedServiceEvents.cs b/src/Dibix.Worker.Abstractions/IHostedServiceEvents.cs
@@ -0,0 +1,11 @@
+﻿using System.Threading;
+using System.Threading.Tasks;
+
+namespace Dibix.Worker.Abstractions
+{
+    public interface IHostedServiceEvents
+    {
+        Task OnWorkerStarted(string fullName, CancellationToken cancellationToken);
+        Task OnServiceBrokerIterationCompleted(string fullName, CancellationToken cancellationToken);
+    }
+}
diff --git a/src/Dibix.Worker.Abstractions/IHostedServiceRegistrar.cs b/src/Dibix.Worker.Abstractions/IHostedServiceRegistrar.cs
diff --git a/src/Dibix.Worker.Abstractions/IWorkerHostExtensionConfigurationBuilder.cs b/src/Dibix.Worker.Abstractions/IWorkerHostExtensionConfigurationBuilder.cs
@@ -12,8 +12,11 @@ public interface IWorkerHostExtensionConfigurationBuilder
         IWorkerHostExtensionConfigurationBuilder ConfigureConnectionString(Func<string?, string?> configure);
         IWorkerHostExtensionConfigurationBuilder OnHostStarted(Func<IWorkerDependencyContext, Task> handler);
         IWorkerHostExtensionConfigurationBuilder OnHostStopped(Func<IWorkerDependencyContext, Task> handler);
-        IWorkerHostExtensionConfigurationBuilder OnWorkerRegistered(OnWorkerRegistered handler);
+        IWorkerHostExtensionConfigurationBuilder OnWorkerStarted(OnWorkerStarted handler);
+        IWorkerHostExtensionConfigurationBuilder OnServiceBrokerIterationCompleted(OnServiceBrokerIterationCompleted handler);
     }
 
-    public delegate Task OnWorkerRegistered(string implementationName, IWorkerDependencyContext dependencyContext, CancellationToken cancellationToken);
+    public delegate Task OnWorkerStarted(string implementationName, IWorkerDependencyContext dependencyContext, CancellationToken cancellationToken);
+    
+    public delegate Task OnServiceBrokerIterationCompleted(string implementationName, IWorkerDependencyContext dependencyContext, CancellationToken cancellationToken);
 }
diff --git a/src/Dibix.Worker.Abstractions/ServiceBrokerSignalSubscriber.cs b/src/Dibix.Worker.Abstractions/ServiceBrokerSignalSubscriber.cs
@@ -8,7 +8,7 @@ namespace Dibix.Worker.Abstractions
 {
     public abstract class ServiceBrokerSignalSubscriber : ServiceBrokerSubscriber<object>, IHostedService
     {
-        protected ServiceBrokerSignalSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(scopeFactory, hostedServiceRegistrar, logger) { }
+        protected ServiceBrokerSignalSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(scopeFactory, hostedServiceEvents, logger) { }
 
         private protected sealed override Task ProcessMessages(IEnumerable<object> messages) => OnMessage();
 
diff --git a/src/Dibix.Worker.Abstractions/ServiceBrokerSubscriber.cs b/src/Dibix.Worker.Abstractions/ServiceBrokerSubscriber.cs
@@ -19,16 +19,18 @@ public abstract class ServiceBrokerSubscriber<TMessage> : BackgroundService, IHo
         private const int ReceiveTimeout    = CommandTimeout / 2 * 1000; // ms
         private const int RetryOnErrorDelay = 10000;                     // ms
         private readonly IWorkerScopeFactory _scopeFactory;
+        private readonly IHostedServiceEvents _hostedServiceEvents;
         #endregion
 
         #region Properties
         protected abstract string ReceiveProcedureName { get; }
         #endregion
 
         #region Constructor
-        protected ServiceBrokerSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(hostedServiceRegistrar, logger)
+        protected ServiceBrokerSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(hostedServiceEvents, logger)
         {
             _scopeFactory = scopeFactory;
+            _hostedServiceEvents = hostedServiceEvents;
         }
         #endregion
 
@@ -41,6 +43,7 @@ protected sealed override async Task ExecuteAsync(CancellationToken stoppingToke
                 try
                 {
                     ICollection<TMessage> messages = await ReceiveMessageBatch(stoppingToken).ConfigureAwait(false);
+                    _ = _hostedServiceEvents.OnServiceBrokerIterationCompleted(GetType().FullName, stoppingToken);
                     if (messages.Any())
                         _ = ProcessMessages(messages);
                 }
diff --git a/src/Dibix.Worker.Host/Configuration/HostedServiceEventOptions.cs b/src/Dibix.Worker.Host/Configuration/HostedServiceEventOptions.cs
@@ -0,0 +1,10 @@
+﻿using Dibix.Worker.Abstractions;
+
+namespace Dibix.Worker.Host
+{
+    public sealed class HostedServiceEventOptions
+    {
+        internal OnWorkerStarted? OnWorkerStarted { get; set; }
+        internal OnServiceBrokerIterationCompleted? OnServiceBrokerIterationCompleted { get; set; }
+    }
+}
diff --git a/src/Dibix.Worker.Host/Configuration/HostedServiceRegistrarOptions.cs b/src/Dibix.Worker.Host/Configuration/HostedServiceRegistrarOptions.cs
diff --git a/src/Dibix.Worker.Host/Program.cs b/src/Dibix.Worker.Host/Program.cs
@@ -31,7 +31,7 @@ private static async Task Main(string[] args)
                     .AddScoped<IDatabaseAccessorFactory, ScopedDatabaseAccessorFactory>()
                     .AddScoped<IWorkerDependencyContext, ServiceProviderWorkerDependencyContext>()
                     .AddSingleton<IWorkerScopeFactory, ServiceScopeWorkerScopeFactory>()
-                    .AddSingleton<IHostedServiceRegistrar, DefaultHostedServiceRegistrar>()
+                    .AddSingleton<IHostedServiceEvents, HostedServiceEvents>()
                     .AddHostedService<DatabaseOptionsMonitor>();
 
             services.Configure<DatabaseOptions>(builder.Configuration.GetSection(DatabaseOptions.ConfigurationSectionName));
diff --git a/src/Dibix.Worker.Host/Registration/DefaultHostedServiceRegistrar.cs b/src/Dibix.Worker.Host/Registration/DefaultHostedServiceRegistrar.cs
diff --git a/src/Dibix.Worker.Host/Registration/HostedServiceEvents.cs b/src/Dibix.Worker.Host/Registration/HostedServiceEvents.cs
@@ -0,0 +1,37 @@
+﻿using System.Threading;
+using System.Threading.Tasks;
+using Dibix.Worker.Abstractions;
+using Microsoft.Extensions.Options;
+
+namespace Dibix.Worker.Host
+{
+    internal sealed class HostedServiceEvents : IHostedServiceEvents
+    {
+        private readonly IWorkerScopeFactory _scopeFactory;
+        private readonly IOptions<HostedServiceEventOptions> _options;
+
+        public HostedServiceEvents(IWorkerScopeFactory scopeFactory, IOptions<HostedServiceEventOptions> options)
+        {
+            _scopeFactory = scopeFactory;
+            _options = options;
+        }
+
+        public async Task OnWorkerStarted(string fullName, CancellationToken cancellationToken)
+        {
+            if (_options.Value.OnWorkerStarted == null)
+                return;
+
+            using IWorkerScope scope = _scopeFactory.Create();
+            await _options.Value.OnWorkerStarted(fullName, scope, cancellationToken).ConfigureAwait(false);
+        }
+
+        public async Task OnServiceBrokerIterationCompleted(string fullName, CancellationToken cancellationToken)
+        {
+            if (_options.Value.OnServiceBrokerIterationCompleted == null)
+                return;
+
+            using IWorkerScope scope = _scopeFactory.Create();
+            await _options.Value.OnServiceBrokerIterationCompleted(fullName, scope, cancellationToken).ConfigureAwait(false);
+        }
+    }
+}
diff --git a/src/Dibix.Worker.Host/Registration/WorkerHostExtensionRegistrar.cs b/src/Dibix.Worker.Host/Registration/WorkerHostExtensionRegistrar.cs
@@ -86,9 +86,15 @@ IWorkerHostExtensionConfigurationBuilder IWorkerHostExtensionConfigurationBuilde
                 return this;
             }
 
-            IWorkerHostExtensionConfigurationBuilder IWorkerHostExtensionConfigurationBuilder.OnWorkerRegistered(OnWorkerRegistered handler)
+            IWorkerHostExtensionConfigurationBuilder IWorkerHostExtensionConfigurationBuilder.OnWorkerStarted(OnWorkerStarted handler)
             {
-                _services.Configure<HostedServiceRegistrarOptions>(x => x.Handler = handler);
+                _services.Configure<HostedServiceEventOptions>(x => x.OnWorkerStarted = handler);
+                return this;
+            }
+
+            IWorkerHostExtensionConfigurationBuilder IWorkerHostExtensionConfigurationBuilder.OnServiceBrokerIterationCompleted(OnServiceBrokerIterationCompleted handler)
+            {
+                _services.Configure<HostedServiceEventOptions>(x => x.OnServiceBrokerIterationCompleted = handler);
                 return this;
             }
 

Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,7 @@ public abstract class BackgroundService : HostedServiceListener, IHostedService,`
`11`	`11`	`private Task? _executingTask;`
`12`	`12`	`private CancellationTokenSource? _stoppingCts;`
`13`	`13`
`14`		`- protected BackgroundService(IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(hostedServiceRegistrar, logger) { }`
	`14`	`+ protected BackgroundService(IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(hostedServiceEvents, logger) { }`
`15`	`15`
`16`	`16`	`protected sealed override Task StartListenerAsync(CancellationToken cancellationToken)`
`17`	`17`	`{`
Original file line number	Diff line number	Diff line change
`@@ -7,16 +7,16 @@ namespace Dibix.Worker.Abstractions`
`7`	`7`	`{`
`8`	`8`	`public abstract class HostedServiceListener : HostedService, IHostedService`
`9`	`9`	`{`
`10`		`- private readonly IHostedServiceRegistrar _hostedServiceRegistrar;`
	`10`	`+ private readonly IHostedServiceEvents _hostedServiceEvents;`
`11`	`11`
`12`		`- protected HostedServiceListener(IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(logger)`
	`12`	`+ protected HostedServiceListener(IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(logger)`
`13`	`13`	`{`
`14`		`- _hostedServiceRegistrar = hostedServiceRegistrar;`
	`14`	`+ _hostedServiceEvents = hostedServiceEvents;`
`15`	`15`	`}`
`16`	`16`
`17`	`17`	`protected sealed override async Task StartServiceAsync(CancellationToken cancellationToken)`
`18`	`18`	`{`
`19`		`- await _hostedServiceRegistrar.RegisterHostedService(GetType().FullName, cancellationToken).ConfigureAwait(false);`
	`19`	`+ await _hostedServiceEvents.OnWorkerStarted(GetType().FullName, cancellationToken).ConfigureAwait(false);`
`20`	`20`	`await StartListenerAsync(cancellationToken).ConfigureAwait(false);`
`21`	`21`	`}`
`22`	`22`
Original file line number	Diff line number	Diff line change
`@@ -12,8 +12,11 @@ public interface IWorkerHostExtensionConfigurationBuilder`
`12`	`12`	`IWorkerHostExtensionConfigurationBuilder ConfigureConnectionString(Func<string?, string?> configure);`
`13`	`13`	`IWorkerHostExtensionConfigurationBuilder OnHostStarted(Func<IWorkerDependencyContext, Task> handler);`
`14`	`14`	`IWorkerHostExtensionConfigurationBuilder OnHostStopped(Func<IWorkerDependencyContext, Task> handler);`
`15`		`- IWorkerHostExtensionConfigurationBuilder OnWorkerRegistered(OnWorkerRegistered handler);`
	`15`	`+ IWorkerHostExtensionConfigurationBuilder OnWorkerStarted(OnWorkerStarted handler);`
	`16`	`+ IWorkerHostExtensionConfigurationBuilder OnServiceBrokerIterationCompleted(OnServiceBrokerIterationCompleted handler);`
`16`	`17`	`}`
`17`	`18`
`18`		`- public delegate Task OnWorkerRegistered(string implementationName, IWorkerDependencyContext dependencyContext, CancellationToken cancellationToken);`
	`19`	`+ public delegate Task OnWorkerStarted(string implementationName, IWorkerDependencyContext dependencyContext, CancellationToken cancellationToken);`
	`20`	`+`
	`21`	`+ public delegate Task OnServiceBrokerIterationCompleted(string implementationName, IWorkerDependencyContext dependencyContext, CancellationToken cancellationToken);`
`19`	`22`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,7 +8,7 @@ namespace Dibix.Worker.Abstractions`
`8`	`8`	`{`
`9`	`9`	`public abstract class ServiceBrokerSignalSubscriber : ServiceBrokerSubscriber<object>, IHostedService`
`10`	`10`	`{`
`11`		`- protected ServiceBrokerSignalSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(scopeFactory, hostedServiceRegistrar, logger) { }`
	`11`	`+ protected ServiceBrokerSignalSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(scopeFactory, hostedServiceEvents, logger) { }`
`12`	`12`
`13`	`13`	`private protected sealed override Task ProcessMessages(IEnumerable<object> messages) => OnMessage();`
`14`	`14`
Original file line number	Diff line number	Diff line change
`@@ -19,16 +19,18 @@ public abstract class ServiceBrokerSubscriber<TMessage> : BackgroundService, IHo`
`19`	`19`	`private const int ReceiveTimeout = CommandTimeout / 2 * 1000; // ms`
`20`	`20`	`private const int RetryOnErrorDelay = 10000; // ms`
`21`	`21`	`private readonly IWorkerScopeFactory _scopeFactory;`
	`22`	`+ private readonly IHostedServiceEvents _hostedServiceEvents;`
`22`	`23`	`#endregion`
`23`	`24`
`24`	`25`	`#region Properties`
`25`	`26`	`protected abstract string ReceiveProcedureName { get; }`
`26`	`27`	`#endregion`
`27`	`28`
`28`	`29`	`#region Constructor`
`29`		`- protected ServiceBrokerSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceRegistrar hostedServiceRegistrar, ILogger logger) : base(hostedServiceRegistrar, logger)`
	`30`	`+ protected ServiceBrokerSubscriber(IWorkerScopeFactory scopeFactory, IHostedServiceEvents hostedServiceEvents, ILogger logger) : base(hostedServiceEvents, logger)`
`30`	`31`	`{`
`31`	`32`	`_scopeFactory = scopeFactory;`
	`33`	`+ _hostedServiceEvents = hostedServiceEvents;`
`32`	`34`	`}`
`33`	`35`	`#endregion`
`34`	`36`
`@@ -41,6 +43,7 @@ protected sealed override async Task ExecuteAsync(CancellationToken stoppingToke`
`41`	`43`	`try`
`42`	`44`	`{`
`43`	`45`	`ICollection<TMessage> messages = await ReceiveMessageBatch(stoppingToken).ConfigureAwait(false);`
	`46`	`+ _ = _hostedServiceEvents.OnServiceBrokerIterationCompleted(GetType().FullName, stoppingToken);`
`44`	`47`	`if (messages.Any())`
`45`	`48`	`_ = ProcessMessages(messages);`
`46`	`49`	`}`