Update event processor event type adding behaviour to use delegates

Author: farajfarook

.github/workflows/cd.yaml

@@ -6,7 +6,7 @@ on:
       - main
 
 env:
-  PACKAGE_VERSION: 6.2.1
+  PACKAGE_VERSION: 6.3.0
   PACKAGE_PROJECT: https://nlib.enbiso.com
   PACKAGE_REPO: https://github.com/enbiso/Enbiso.NLib
   PACKAGE_COPYRIGHT: Copyright 2021 (c) enbiso. All rights reserved.

Enbiso.NLib.EventBus.Abstractions/IEventHandler.cs

@@ -8,7 +8,8 @@ namespace Enbiso.NLib.EventBus
     /// </summary>
     public interface IEventHandler
     {
-        Type GetEventType();
+        string EventName { get; }
+        Type EventType { get; }
         Task Handle(object @event);
     }
 
@@ -19,7 +20,8 @@ public interface IEventHandler
     public abstract class EventHandler<TEvent>: IEventHandler where TEvent : class, IEvent
     {
         protected abstract Task Handle(TEvent @event);
-        public Type GetEventType() => typeof(TEvent);
+        public string EventName => typeof(TEvent).Name;
+        public Type EventType => typeof(TEvent);
         public Task Handle(object @event) => Handle(@event as TEvent);
     }
 }

Enbiso.NLib.EventBus.Nats/JetStream/JetStreamEventSubscriber.cs

@@ -20,7 +20,6 @@ public JetStreamEventSubscriber(IJetStreamConnection jsConnection, IOptions<Nats
             _eventProcessor = eventProcessor;
             _natsConnection = natsConnection;
             _options = options.Value;
-            _eventProcessor.Setup();
         }
 
         public void Dispose()

Enbiso.NLib.EventBus.Nats/NatsEventSubscriber.cs

@@ -16,7 +16,6 @@ public NatsEventSubscriber(INatsConnection connection, IOptions<NatsOptions> opt
         _connection = connection;
         _eventProcessor = eventProcessor;
         _options = options.Value;
-        _eventProcessor.Setup();
     }
 
     public void Dispose()

Enbiso.NLib.EventBus.RabbitMq/RabbitMqBusSubscriber.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -19,17 +20,17 @@ public RabbitMqBusSubscriber(IRabbitMqConnection connection, IEventProcessor eve
         {
             var option = optionWrap.Value;
             _queueName = option.Client;
-            _exchanges = option.Exchanges?? new string[0];
+            _exchanges = option.Exchanges?? Array.Empty<string>();
 
             _connection = connection;
             _eventProcessor = eventProcessor;
 
-            _eventProcessor.Setup(eventName =>
+            _eventProcessor.EventTypeAdded += (_, args) =>
             {
                 using var channel = _connection.CreateModel();
                 foreach (var exchange in _exchanges)
-                    channel.QueueBind(queue: _queueName, exchange: exchange, routingKey: eventName);
-            });
+                    channel.QueueBind(queue: _queueName, exchange: exchange, routingKey: args.EventName);
+            };
         }
 
         private IModel CreateConsumerChannel()

Enbiso.NLib.EventBus/EventProcessor.cs

@@ -10,13 +10,14 @@ namespace Enbiso.NLib.EventBus
     public class EventProcessor: IEventProcessor
     {
         private readonly Dictionary<string, List<IEventHandler>> _subscriptions = new();
-        
-        private readonly IEnumerable<IEventHandler> _eventHandlers;
         private readonly ILogger _logger;
-        
+
         public EventProcessor(IEnumerable<IEventHandler> eventHandlers, ILogger<EventProcessor> logger)
         {
-            _eventHandlers = eventHandlers;
+            foreach (var eventHandler in eventHandlers)
+            {
+                AddEventHandler(eventHandler);
+            }
             _logger = logger;
         }
 
@@ -29,7 +30,7 @@ public async Task ProcessEvent(string eventName, byte[] data)
             var eventHandlers = _subscriptions[eventName];
             foreach (var eventHandler in eventHandlers)
             {
-                var eventType = eventHandler.GetEventType();
+                var eventType = eventHandler.EventType;
                 var @event = JsonSerializer.Deserialize(message, eventType);
 
                 var eventId = @event is IEvent iEvent ? iEvent.EventId : Guid.Empty;
@@ -45,23 +46,23 @@ public async Task ProcessEvent(string eventName, byte[] data)
             }
         }
 
-        public void Setup(Action<string> onAddSubscription)
+        public event EventProcessorEventTypeAddedEventHandler EventTypeAdded;
+        public void AddEventHandler(IEventHandler eventHandler)
         {
-            foreach (var eventHandler in _eventHandlers)
+            var eventName = eventHandler.EventName;
+            if (_subscriptions.TryGetValue(eventName, out var currentHandlers))
             {
-                var eventName = eventHandler.GetEventType().Name;
-                
-                onAddSubscription?.Invoke(eventName);
-                
-                if (_subscriptions.TryGetValue(eventName, out var currentHandlers))
-                {
-                    currentHandlers.Add(eventHandler);
-                    _subscriptions[eventName] = currentHandlers;
-                }
-                else
+                currentHandlers.Add(eventHandler);
+                _subscriptions[eventName] = currentHandlers;
+            }
+            else
+            {
+                _subscriptions.Add(eventName, new List<IEventHandler> {eventHandler});
+                EventTypeAdded?.Invoke(this, new EventProcessorEventTypeAddedEventArgs
                 {
-                    _subscriptions.Add(eventName, new List<IEventHandler> {eventHandler});
-                }
+                    EventType = eventHandler.EventType,
+                    EventName = eventName
+                });
             }
         }
     }

Enbiso.NLib.EventBus/EventSubscriptionService.cs

@@ -14,7 +14,7 @@ public EventSubscriptionService(IEnumerable<IEventSubscriber> subscribers)
             _subscribers = subscribers;
         }
 
-        public Task SubscribeAll(CancellationToken token) =>
+        public Task SubscribeAll(CancellationToken token = default) =>
             Task.WhenAll(_subscribers.Select(s => s.Subscribe(token)));
     }
 }

Enbiso.NLib.EventBus/IEventProcessor.cs

@@ -6,6 +6,16 @@ namespace Enbiso.NLib.EventBus
     public interface IEventProcessor
     {
         Task ProcessEvent(string eventName, byte[] data);
-        void Setup(Action<string> onAddSubscription = null);
+        void AddEventHandler(IEventHandler handler);
+        event EventProcessorEventTypeAddedEventHandler EventTypeAdded;
     }
+    
+    public class EventProcessorEventTypeAddedEventArgs: EventArgs
+    {
+        public string EventName { get; set; }
+        public Type EventType { get; set; }
+    }
+    
+    public delegate void EventProcessorEventTypeAddedEventHandler(object sender,
+        EventProcessorEventTypeAddedEventArgs e);
 }