feat: add event notification feature

src/KafkaFlow.Abstractions/Consumers/IWorker.cs

@@ -1,8 +1,6 @@
 namespace KafkaFlow
 {
     using System;
-    using KafkaFlow.Configuration;
-    using KafkaFlow.Observer;
 
     /// <summary>
     /// Represents the interface of a internal worker
@@ -23,11 +21,11 @@ public interface IWorker
         /// <summary>
         /// Gets the subject for worker stopping events where observers can subscribe to receive notifications.
         /// </summary>
-        ISubject<WorkerStoppingSubject, VoidObject> WorkerStopping { get; }
+        IEvent WorkerStopping { get; }
 
         /// <summary>
         /// Gets the subject for worker stopped events where observers can subscribe to receive notifications.
         /// </summary>
-        ISubject<WorkerStoppedSubject, VoidObject> WorkerStopped { get; }
+        IEvent WorkerStopped { get; }
     }
 }

src/KafkaFlow.Abstractions/IEvent.cs

@@ -0,0 +1,15 @@
+namespace KafkaFlow
+{
+    using System;
+    using System.Threading.Tasks;
+
+    public interface IEvent
+    {
+        void Subscribe(Func<Task> handle);
+    }
+
+    public interface IEvent<TArg>
+    {
+        void Subscribe(Func<TArg, Task> handle);
+    }
+}

src/KafkaFlow.Admin.Dashboard/Properties/launchSettings.json

@@ -0,0 +1,12 @@
+{
+  "profiles": {
+    "KafkaFlow.Admin.Dashboard": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      },
+      "applicationUrl": "https://localhost:53859;http://localhost:53860"
+    }
+  }
+}

src/KafkaFlow.BatchConsume/BatchConsumeMiddleware.cs

@@ -7,12 +7,8 @@
     using System.Threading.Tasks;
     using KafkaFlow.Configuration;
     using KafkaFlow.Consumers;
-    using KafkaFlow.Observer;
 
-    internal class BatchConsumeMiddleware
-        : IMessageMiddleware,
-            ISubjectObserver<WorkerStoppedSubject, VoidObject>,
-            IDisposable
+    internal class BatchConsumeMiddleware : IMessageMiddleware, IDisposable
     {
         private readonly SemaphoreSlim dispatchSemaphore = new(1, 1);
 
@@ -37,7 +33,7 @@ public BatchConsumeMiddleware(
             this.batch = new(batchSize);
             this.consumerConfiguration = middlewareContext.Consumer.Configuration;
 
-            middlewareContext.Worker.WorkerStopped.Subscribe(this);
+            middlewareContext.Worker.WorkerStopped.Subscribe(() => this.TriggerDispatchAndWaitAsync());
         }
 
         public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
@@ -67,8 +63,6 @@ public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
             }
         }
 
-        public async Task OnNotification(WorkerStoppedSubject subject, VoidObject arg) => await this.TriggerDispatchAndWaitAsync();
-
         public void Dispose()
         {
             this.dispatchTask?.Dispose();

src/KafkaFlow.UnitTests/BatchConsume/BatchConsumeMiddlewareTests.cs

@@ -51,7 +51,7 @@ public void Setup()
 
             workerMock
                 .SetupGet(x => x.WorkerStopped)
-                .Returns(new WorkerStoppedSubject(this.logHandlerMock.Object));
+                .Returns(new Event(this.logHandlerMock.Object));
 
             consumerConfigurationMock
                 .SetupGet(x => x.AutoMessageCompletion)

src/KafkaFlow.UnitTests/EventTests.cs

@@ -0,0 +1,87 @@
+namespace KafkaFlow.UnitTests
+{
+    using System.Threading.Tasks;
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using Moq;
+
+    [TestClass]
+    public class EventTests
+    {
+        [TestMethod]
+        public async Task FireAsync_EventSubscribed_CallDelegateWithSuccess()
+        {
+            // Arrange
+            var numberOfCalls = 0;
+            var log = new Mock<ILogHandler>();
+
+            var event1 = new Event(log.Object);
+
+            event1.Subscribe(() =>
+            {
+                numberOfCalls++;
+                return Task.CompletedTask;
+            });
+
+            // Act
+            await event1.FireAsync();
+
+            // Assert
+            Assert.AreEqual(1, numberOfCalls);
+        }
+
+        [TestMethod]
+        public async Task FireAsync_EventWithMultipleObservers_CallAllDelegatesWithSuccess()
+        {
+            // Arrange
+            var numberOfCalls = 0;
+            var log = new Mock<ILogHandler>();
+
+            var event1 = new Event(log.Object);
+
+            event1.Subscribe(() =>
+            {
+                numberOfCalls++;
+                return Task.CompletedTask;
+            });
+
+            event1.Subscribe(() =>
+            {
+                numberOfCalls++;
+                return Task.CompletedTask;
+            });
+
+            // Act
+            await event1.FireAsync();
+
+            // Assert
+            Assert.AreEqual(2, numberOfCalls);
+        }
+
+        [TestMethod]
+        public async Task FireAsync_EventWithMultipleObserversAndErrors_CallAllDelegatesAndContinueWithoutErrors()
+        {
+            // Arrange
+            var numberOfCalls = 0;
+            var log = new Mock<ILogHandler>();
+
+            var event1 = new Event(log.Object);
+
+            event1.Subscribe(() =>
+            {
+                throw new System.Exception();
+            });
+
+            event1.Subscribe(() =>
+            {
+                numberOfCalls++;
+                return Task.CompletedTask;
+            });
+
+            // Act
+            await event1.FireAsync();
+
+            // Assert
+            Assert.AreEqual(1, numberOfCalls);
+        }
+    }
+}

src/KafkaFlow/ConsumerManagerFactory.cs

@@ -22,7 +22,7 @@ public IConsumerManager Create(IConsumerConfiguration configuration, IDependency
                 configuration,
                 logHandler);
 
-            consumerWorkerPool.WorkerPoolStopped.Subscribe(middlewareExecutor);
+            consumerWorkerPool.WorkerPoolStopped.Subscribe(() => middlewareExecutor.OnNotification());
 
             var feeder = new WorkerPoolFeeder(
                 consumer,

src/KafkaFlow/Consumers/ConsumerWorker.cs

@@ -4,7 +4,6 @@ namespace KafkaFlow.Consumers
     using System.Threading;
     using System.Threading.Channels;
     using System.Threading.Tasks;
-    using KafkaFlow.Observer;
 
     internal class ConsumerWorker : IConsumerWorker
     {
@@ -15,8 +14,8 @@ internal class ConsumerWorker : IConsumerWorker
 
         private readonly Channel<IMessageContext> messagesBuffer;
 
-        private readonly WorkerStoppingSubject workerStoppingSubject;
-        private readonly WorkerStoppedSubject workerStoppedSubject;
+        private readonly Event workerStoppingSubject;
+        private readonly Event workerStoppedSubject;
 
         private CancellationTokenSource stopCancellationTokenSource;
         private Task backgroundTask;
@@ -51,9 +50,9 @@ public ConsumerWorker(
 
         public IDependencyResolver WorkerDependencyResolver => this.workerDependencyResolverScope.Resolver;
 
-        public ISubject<WorkerStoppingSubject, VoidObject> WorkerStopping => this.workerStoppingSubject;
+        public IEvent WorkerStopping => this.workerStoppingSubject;
 
-        public ISubject<WorkerStoppedSubject, VoidObject> WorkerStopped => this.workerStoppedSubject;
+        public IEvent WorkerStopped => this.workerStoppedSubject;
 
         public ValueTask EnqueueAsync(
             IMessageContext context,
@@ -98,7 +97,7 @@ public Task StartAsync()
 
         public async Task StopAsync()
         {
-            await this.workerStoppingSubject.NotifyAsync(VoidObject.Value);
+            await this.workerStoppingSubject.FireAsync();
 
             this.messagesBuffer.Writer.TryComplete();
 
@@ -109,7 +108,7 @@ public async Task StopAsync()
 
             await this.backgroundTask.ConfigureAwait(false);
 
-            await this.workerStoppedSubject.NotifyAsync(VoidObject.Value);
+            await this.workerStoppedSubject.FireAsync();
         }
 
         public void Dispose()