Features/observer pattern

src/KafkaFlow.Abstractions/Consumers/IWorker.cs

@@ -1,7 +1,6 @@
 namespace KafkaFlow
 {
     using System;
-    using KafkaFlow.Observer;
 
     /// <summary>
     /// Represents the interface of a internal worker
@@ -14,19 +13,18 @@ public interface IWorker
         int Id { get; }
 
         /// <summary>
-        /// This handler is called immediately after a worker completes the consumption of a message
+        /// Gets the subject for worker stopping events where observers can subscribe to receive notifications.
         /// </summary>
-        /// <param name="handler"><see cref="Action"/> to be executed</param>
-        void OnTaskCompleted(Action handler);
+        IEvent WorkerStopping { get; }
 
         /// <summary>
-        /// Gets the subject for worker stopping events where observers can subscribe to receive notifications.
+        /// Gets the subject for worker stopped events where observers can subscribe to receive notifications.
         /// </summary>
-        ISubject<WorkerStoppingSubject, VoidObject> WorkerStopping { get; }
+        IEvent WorkerStopped { get; }
 
         /// <summary>
-        /// Gets the subject for worker stopped events where observers can subscribe to receive notifications.
+        /// Gets the subject for worker consumption completed events where observers can subscribe to receive notifications.
         /// </summary>
-        ISubject<WorkerStoppedSubject, VoidObject> WorkerStopped { get; }
+        IEvent<IMessageContext> TaskCompleted { get; }
     }
 }

src/KafkaFlow.Abstractions/IEvent.cs

@@ -0,0 +1,32 @@
+namespace KafkaFlow
+{
+    using System;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Represents an Event to be subscribed.
+    /// </summary>
+    public interface IEvent
+    {
+        /// <summary>
+        /// Subscribes to the event.
+        /// </summary>
+        /// <param name="handler">The handler to be called when the event is fired.</param>
+        /// <returns>Event subscription reference</returns>
+        IEventSubscription Subscribe(Func<Task> handler);
+    }
+
+    /// <summary>
+    /// Represents an Event to be subscribed.
+    /// </summary>
+    /// <typeparam name="TArg">The argument expected by the event.</typeparam>
+    public interface IEvent<TArg>
+    {
+        /// <summary>
+        /// Subscribes to the event.
+        /// </summary>
+        /// <param name="handler">The handler to be called when the event is fired.</param>
+        /// <returns>Event subscription reference</returns>
+        IEventSubscription Subscribe(Func<TArg, Task> handler);
+    }
+}

src/KafkaFlow.Abstractions/IEventSubscription.cs

@@ -0,0 +1,15 @@
+using System;
+
+namespace KafkaFlow
+{
+    /// <summary>
+    /// Represents an Event subscription.
+    /// </summary>
+    public interface IEventSubscription
+    {
+        /// <summary>
+        /// Cancels the subscription to the event.
+        /// </summary>
+        void Cancel();
+    }
+}

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)