Merge pull request #14 from alifcapital/mirolimmajidov-acm1-361

Improve activity log when it is creating from RabbitMQ event (EventBus.RabbitMQ part)

src/EventBus.RabbitMQ.csproj

@@ -18,7 +18,7 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="AlifCapital.EventStorage" Version="9.0.6" />
+        <PackageReference Include="AlifCapital.EventStorage" Version="9.0.8" />
         <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="9.0.0" />
         <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.0" />
         <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.0" />

src/Extensions/RabbitMqExtensions.cs

@@ -10,6 +10,7 @@
 using EventBus.RabbitMQ.Subscribers.Options;
 using EventStorage.Configurations;
 using EventStorage.Extensions;
+using EventStorage.Inbox.EventArgs;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 
@@ -28,19 +29,33 @@ public static class RabbitMqExtensions
     /// <param name="eventSubscriberManagerOptions">Options to register subscriber with the settings. It will overwrite existing subscriber setting if exists</param>
     /// <param name="eventStoreOptions">Options to overwrite default settings of Inbox and Outbox.</param>
     /// <param name="assemblies">Assemblies to find and load publisher and subscribers</param>
+    /// <param name="executingSubscribedEvent">The event for subscribing to the executing subscribed event of MessageBroker</param>
+    /// <param name="executingReceivedEvent">The event for subscribing to the executing received event of Inbox</param>
     public static void AddRabbitMqEventBus(this IServiceCollection services, IConfiguration configuration,
         Assembly[] assemblies,
         Action<RabbitMqOptions> defaultOptions = null,
         Action<Dictionary<string, RabbitMqHostSettings>> virtualHostSettingsOptions = null,
         Action<EventPublisherManagerOptions> eventPublisherManagerOptions = null,
         Action<EventSubscriberManagerOptions> eventSubscriberManagerOptions = null,
-        Action<InboxAndOutboxOptions> eventStoreOptions = null)
+        Action<InboxAndOutboxOptions> eventStoreOptions = null,
+        EventHandler<SubscribedMessageBrokerEventArgs> executingSubscribedEvent = null,
+        EventHandler<ReceivedEventArgs> executingReceivedEvent = null)
     {
-        services.AddEventStore(configuration, assemblies: assemblies, options: eventStoreOptions);
+        var eventsToSubscribe = new List<EventHandler<ReceivedEventArgs>>();
+        if (executingReceivedEvent is not null)
+            eventsToSubscribe.Add(executingReceivedEvent);
+
+        if (executingSubscribedEvent is not null)
+        {
+            EventSubscriberManager.ExecutingSubscribedEvent += executingSubscribedEvent;
+            eventsToSubscribe.Add(EventSubscriberManager.HandleExecutingReceivedEvent);
+        }
+        services.AddEventStore(configuration, assemblies: assemblies, options: eventStoreOptions, executingReceivedEvents: eventsToSubscribe.ToArray());
 
         var settings = configuration.GetSection(nameof(RabbitMqSettings)).Get<RabbitMqSettings>() ??
                        new RabbitMqSettings();
         LoadDefaultRabbitMqOptions(settings, defaultOptions);
+
         if (!settings.DefaultSettings.IsEnabled) return;
 
         services.AddSingleton(settings.DefaultSettings);
@@ -144,14 +159,15 @@ private static void RegisterAllPublishers(EventPublisherManager publisherManager
     /// </summary>
     /// <param name="settingsFromConfig">Main settings from configuration</param>
     /// <param name="defaultOptions">Settings option to overwrite the default settings</param>
-    private static void LoadDefaultRabbitMqOptions(RabbitMqSettings settingsFromConfig, Action<RabbitMqOptions> defaultOptions)
+    private static void LoadDefaultRabbitMqOptions(RabbitMqSettings settingsFromConfig,
+        Action<RabbitMqOptions> defaultOptions)
     {
         var defaultSettings = RabbitMqOptionsConstant.CreateDefaultRabbitMqOptions();
         if (settingsFromConfig.DefaultSettings is null)
             settingsFromConfig.DefaultSettings = defaultSettings;
         else
             settingsFromConfig.DefaultSettings.CopyNotAssignedSettingsFrom(defaultSettings);
-        
+
         defaultOptions?.Invoke(settingsFromConfig.DefaultSettings);
     }
 
@@ -173,13 +189,14 @@ private static void RegisterAllSubscribers(EventSubscriberManager subscriberMana
         }
     }
 
-    private static void RegisterAllSubscriberReceiversToDependencyInjection(IServiceCollection services, Assembly[] assemblies)
+    private static void RegisterAllSubscriberReceiversToDependencyInjection(IServiceCollection services,
+        Assembly[] assemblies)
     {
         var subscriberReceiverTypes = GetSubscriberReceiverTypes(assemblies);
         foreach (var (_, handlerType) in subscriberReceiverTypes)
             services.AddTransient(handlerType);
     }
-    
+
     static readonly Type SubscriberReceiverType = typeof(IEventSubscriber<>);
 
     internal static List<(Type eventType, Type handlerType)> GetSubscriberReceiverTypes(Assembly[] assemblies)

src/Subscribers/Consumers/EventConsumerService.cs

@@ -3,6 +3,7 @@
 using System.Text.Json;
 using EventBus.RabbitMQ.Connections;
 using EventBus.RabbitMQ.Instrumentation.Trace;
+using EventBus.RabbitMQ.Subscribers.Managers;
 using EventBus.RabbitMQ.Subscribers.Models;
 using EventBus.RabbitMQ.Subscribers.Options;
 using EventStorage.Exceptions;
@@ -138,7 +139,8 @@ private async Task Consumer_Received(object sender, BasicDeliverEventArgs eventA
                 _logger.LogTrace("Received RabbitMQ event, Type is {EventType} and EventId is {EventId}", eventType,
                     eventArgs.BasicProperties.MessageId);
                 var eventId = Guid.TryParse(eventArgs.BasicProperties.MessageId, out Guid messageId)
-                    ? messageId : Guid.NewGuid();
+                    ? messageId
+                    : Guid.NewGuid();
 
                 using var scope = _serviceProvider.CreateScope();
                 if (_useInbox)
@@ -150,16 +152,21 @@ private async Task Consumer_Received(object sender, BasicDeliverEventArgs eventA
                             $"The RabbitMQ is configured to use the Inbox for received events, but the Inbox functionality of the EventStorage is not enabled. So, the {info.eventHandlerType.Name} event subscriber of an event will be executed immediately for the event id: {eventId};");
 
                     _ = eventReceiverManager.Received(eventId, info.eventType.Name, eventArgs.RoutingKey,
-                        EventProviderType.MessageBroker, payload: eventPayload, headers: headersAsJson, namingPolicyType: info.eventSettings.PropertyNamingPolicy ?? NamingPolicyType.PascalCase);
+                        EventProviderType.MessageBroker, payload: eventPayload, headers: headersAsJson,
+                        namingPolicyType: info.eventSettings.PropertyNamingPolicy ?? NamingPolicyType.PascalCase);
                 }
                 else
                 {
                     var jsonSerializerSetting = info.eventSettings.GetJsonSerializer();
                     var receivedEvent =
-                        JsonSerializer.Deserialize(eventPayload, info.eventType, jsonSerializerSetting) as ISubscribeEvent;
+                        JsonSerializer.Deserialize(eventPayload, info.eventType, jsonSerializerSetting) as
+                            ISubscribeEvent;
 
                     receivedEvent!.EventId = eventId;
                     receivedEvent!.Headers = headers;
+
+                    EventSubscriberManager.OnExecutingSubscribedEvent(receivedEvent, _connectionOptions.VirtualHostSettings.VirtualHost);
+
                     var eventHandlerSubscriber = scope.ServiceProvider.GetRequiredService(info.eventHandlerType);
                     var handleMethod = info.eventHandlerType.GetMethod(HandlerMethodName);
                     await ((Task)handleMethod!.Invoke(eventHandlerSubscriber, [receivedEvent]))!;

src/Subscribers/Managers/EventSubscriberManager.cs

@@ -2,18 +2,25 @@
 using EventBus.RabbitMQ.Subscribers.Consumers;
 using EventBus.RabbitMQ.Subscribers.Models;
 using EventBus.RabbitMQ.Subscribers.Options;
+using EventStorage.Inbox.EventArgs;
+using EventStorage.Models;
 using Microsoft.Extensions.DependencyInjection;
 
 namespace EventBus.RabbitMQ.Subscribers.Managers;
 
 internal class EventSubscriberManager(RabbitMqOptions defaultSettings, IServiceProvider serviceProvider)
     : IEventSubscriberManager
 {
+    /// <summary>
+    /// The event to be executed before executing the subscriber of the received event.
+    /// </summary>
+    public static event EventHandler<SubscribedMessageBrokerEventArgs> ExecutingSubscribedEvent;
+
     /// <summary>
     /// Dictionary collection to store all event and event handler information
     /// </summary>
-    private readonly Dictionary<string, (Type eventType, Type eventHandlerType, EventSubscriberOptions eventSettings)>
-        _subscribers = new();
+    private static readonly Dictionary<string, (Type eventType, Type eventHandlerType, EventSubscriberOptions eventSettings)>
+        Subscribers = new();
 
     /// <summary>
     /// List of consumers for each unique a queue for different virtual host 
@@ -25,7 +32,7 @@ public void AddSubscriber<TEvent, TEventHandler>(Action<EventSubscriberOptions>
         where TEventHandler : class, IEventSubscriber<TEvent>
     {
         var eventType = typeof(TEvent);
-        if (_subscribers.TryGetValue(eventType.Name, out var info))
+        if (Subscribers.TryGetValue(eventType.Name, out var info))
         {
             options?.Invoke(info.eventSettings);
         }
@@ -35,7 +42,7 @@ public void AddSubscriber<TEvent, TEventHandler>(Action<EventSubscriberOptions>
             options?.Invoke(settings);
 
             var handlerType = typeof(TEventHandler);
-            _subscribers.Add(eventType.Name, (eventType, handlerType, settings));
+            Subscribers.Add(eventType.Name, (eventType, handlerType, settings));
         }
     }
 
@@ -47,33 +54,37 @@ public void AddSubscriber<TEvent, TEventHandler>(Action<EventSubscriberOptions>
     /// <param name="subscriberSettings">Settings of subscriber</param>
     public void AddSubscriber(Type typeOfSubscriber, Type typeOfHandler, EventSubscriberOptions subscriberSettings)
     {
-        if (_subscribers.TryGetValue(typeOfSubscriber.Name, out var info))
-            _subscribers[typeOfSubscriber.Name] = (info.eventType, info.eventHandlerType, subscriberSettings);
+        if (Subscribers.TryGetValue(typeOfSubscriber.Name, out var info))
+            Subscribers[typeOfSubscriber.Name] = (info.eventType, info.eventHandlerType, subscriberSettings);
         else
-            _subscribers.Add(typeOfSubscriber.Name, (typeOfSubscriber, typeOfHandler, subscriberSettings));
+            Subscribers.Add(typeOfSubscriber.Name, (typeOfSubscriber, typeOfHandler, subscriberSettings));
     }
 
     /// <summary>
     /// Setting the virtual host and other unassigned settings of subscribers
     /// </summary>
     public void SetVirtualHostAndOwnSettingsOfSubscribers(Dictionary<string, RabbitMqHostSettings> virtualHostsSettings)
     {
-        foreach (var (eventTypeName, (_, _, eventSettings)) in _subscribers)
+        foreach (var (eventTypeName, (_, _, eventSettings)) in Subscribers)
         {
-            var virtualHostSettings = string.IsNullOrEmpty(eventSettings.VirtualHostKey) ? defaultSettings : virtualHostsSettings.GetValueOrDefault(eventSettings.VirtualHostKey, defaultSettings);
+            var virtualHostSettings = string.IsNullOrEmpty(eventSettings.VirtualHostKey)
+                ? defaultSettings
+                : virtualHostsSettings.GetValueOrDefault(eventSettings.VirtualHostKey, defaultSettings);
             eventSettings.SetVirtualHostAndUnassignedSettings(virtualHostSettings, eventTypeName);
         }
     }
 
     public void CreateConsumerForEachQueueAndStartReceivingEvents()
     {
         var eventConsumerCreator = serviceProvider.GetRequiredService<IEventConsumerServiceCreator>();
-        foreach (var (_, eventInfo) in _subscribers)
+        foreach (var (_, eventInfo) in Subscribers)
         {
-            var consumerId = $"{eventInfo.eventSettings.VirtualHostSettings.VirtualHost}-{eventInfo.eventSettings.QueueName}";
-            if (!_eventConsumers.TryGetValue(consumerId, value: out IEventConsumerService eventConsumer))
+            var consumerId =
+                $"{eventInfo.eventSettings.VirtualHostSettings.VirtualHost}-{eventInfo.eventSettings.QueueName}";
+            if (!_eventConsumers.TryGetValue(consumerId, value: out var eventConsumer))
             {
-                eventConsumer = eventConsumerCreator.Create(eventInfo.eventSettings, serviceProvider, defaultSettings.UseInbox);
+                eventConsumer =
+                    eventConsumerCreator.Create(eventInfo.eventSettings, serviceProvider, defaultSettings.UseInbox);
                 _eventConsumers.Add(consumerId, eventConsumer);
             }
 
@@ -83,4 +94,43 @@ public void CreateConsumerForEachQueueAndStartReceivingEvents()
         foreach (var consumer in _eventConsumers)
             consumer.Value.StartAndSubscribeReceiver();
     }
+
+    /// <summary>
+    /// Invokes the ExecutingReceivedEvent event to be able to execute the event before the subscriber.
+    /// </summary>
+    /// <param name="event">Executing an event</param>
+    /// <param name="virtualHostName">The name of virtual host to being able to get a system name that the event published by it.</param>
+    public static void OnExecutingSubscribedEvent(ISubscribeEvent @event, string virtualHostName)
+    {
+        if (ExecutingSubscribedEvent is null)
+            return;
+
+        var systemName = virtualHostName.TrimStart('/');
+        var eventArgs = new SubscribedMessageBrokerEventArgs
+        {
+            Event = @event,
+            SystemName = systemName
+        };
+
+        ExecutingSubscribedEvent.Invoke(null, eventArgs);
+    }
+
+    /// <summary>
+    /// For handling the ExecutingReceivedEvent event and execute the ExecutingSubscribedEvent event if the  
+    /// </summary>
+    public static void HandleExecutingReceivedEvent(object sender, ReceivedEventArgs e)
+    {
+        if (e.ProviderType == EventProviderType.MessageBroker)
+        {
+            if (e.Event is not ISubscribeEvent @event)
+                return;
+
+            var eventTypeName = @event.GetType().Name;
+            var virtualHostName = Subscribers.TryGetValue(eventTypeName, out var info)
+                ? info.eventSettings.VirtualHostSettings.VirtualHost
+                : string.Empty;
+
+            OnExecutingSubscribedEvent(@event, virtualHostName);
+        }
+    }
 }

src/Subscribers/Models/SubscribedMessageBrokerEventArgs.cs

@@ -0,0 +1,14 @@
+namespace EventBus.RabbitMQ.Subscribers.Models;
+
+public class SubscribedMessageBrokerEventArgs : EventArgs
+{
+    /// <summary>
+    /// Executing received event.
+    /// </summary>
+    public required ISubscribeEvent Event { get; init; }
+
+    /// <summary>
+    /// The name of system that the event published by it.
+    /// </summary>
+    public required string SystemName { get; init; }
+}