Message Bus

phongnguyend · phongnguyend · commit 61e37dd77ff3 · 2023-09-04T09:19:19.000+07:00
diff --git a/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageBus.cs b/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageBus.cs
@@ -0,0 +1,14 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+
+public interface IMessageBus
+{
+    Task SendAsync<T>(T message, MetaData metaData = null, CancellationToken cancellationToken = default)
+        where T : IMessageBusEvent;
+
+    Task ReceiveAsync<TConsumer, T>(Func<T, MetaData, Task> action, CancellationToken cancellationToken = default)
+        where T : IMessageBusEvent;
+}
diff --git a/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageBusEvent.cs b/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageBusEvent.cs
@@ -0,0 +1,5 @@
+﻿namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+
+public interface IMessageBusEvent
+{
+}
diff --git a/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MessageBus.cs b/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MessageBus.cs
@@ -0,0 +1,28 @@
+﻿using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+
+public class MessageBus : IMessageBus
+{
+    private readonly IServiceProvider _serviceProvider;
+
+    public MessageBus(IServiceProvider serviceProvider)
+    {
+        _serviceProvider = serviceProvider;
+    }
+
+    public async Task SendAsync<T>(T message, MetaData metaData = null, CancellationToken cancellationToken = default)
+        where T : IMessageBusEvent
+    {
+        await _serviceProvider.GetRequiredService<IMessageSender<T>>().SendAsync(message, metaData, cancellationToken);
+    }
+
+    public async Task ReceiveAsync<TConsumer, T>(Func<T, MetaData, Task> action, CancellationToken cancellationToken = default)
+        where T : IMessageBusEvent
+    {
+        await _serviceProvider.GetRequiredService<IMessageReceiver<TConsumer, T>>().ReceiveAsync(action, cancellationToken);
+    }
+}
diff --git a/src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/AuditLogModuleServiceCollectionExtensions.cs b/src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/AuditLogModuleServiceCollectionExtensions.cs
@@ -1,4 +1,5 @@
-﻿using ClassifiedAds.Domain.Repositories;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Domain.Repositories;
 using ClassifiedAds.Services.AuditLog.Authorization;
 using ClassifiedAds.Services.AuditLog.ConfigurationOptions;
 using ClassifiedAds.Services.AuditLog.DTOs;
@@ -29,7 +30,8 @@ public static IServiceCollection AddAuditLogModule(this IServiceCollection servi
 
         services.AddAuthorizationPolicies(Assembly.GetExecutingAssembly(), AuthorizationPolicyNames.GetPolicyNames());
 
-        services.AddMessageBusReceiver<AuditLogAggregationConsumer, AuditLogCreatedEvent>(appSettings.MessageBroker);
+        services.AddTransient<IMessageBus, MessageBus>()
+                .AddMessageBusReceiver<AuditLogAggregationConsumer, AuditLogCreatedEvent>(appSettings.MessageBroker);
 
         return services;
     }
diff --git a/src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/DTOs/AuditLogCreatedEvent.cs b/src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/DTOs/AuditLogCreatedEvent.cs
@@ -1,8 +1,9 @@
-﻿using ClassifiedAds.Services.AuditLog.Entities;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Services.AuditLog.Entities;
 
 namespace ClassifiedAds.Services.AuditLog.DTOs;
 
-public class AuditLogCreatedEvent
+public class AuditLogCreatedEvent : IMessageBusEvent
 {
     public AuditLogEntry AuditLog { get; set; }
 }
diff --git a/src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/HostedServices/MessageBusReceiver.cs b/src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/HostedServices/MessageBusReceiver.cs
@@ -14,20 +14,20 @@ internal class MessageBusReceiver : BackgroundService
 {
     private readonly ILogger<MessageBusReceiver> _logger;
     private readonly IServiceProvider _serviceProvider;
-    private readonly IMessageReceiver<AuditLogAggregationConsumer, AuditLogCreatedEvent> _auditLogCreatedEventReceiver;
+    private readonly IMessageBus _messageBus;
 
     public MessageBusReceiver(ILogger<MessageBusReceiver> logger,
         IServiceProvider serviceProvider,
-        IMessageReceiver<AuditLogAggregationConsumer, AuditLogCreatedEvent> auditLogCreatedEventReceiver)
+        IMessageBus messageBus)
     {
         _logger = logger;
         _serviceProvider = serviceProvider;
-        _auditLogCreatedEventReceiver = auditLogCreatedEventReceiver;
+        _messageBus = messageBus;
     }
 
     protected override Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        _auditLogCreatedEventReceiver?.ReceiveAsync(async (data, metaData) =>
+        _messageBus.ReceiveAsync<AuditLogAggregationConsumer, AuditLogCreatedEvent>(async (data, metaData) =>
         {
             using var scope = _serviceProvider.CreateScope();
             await ProcessMessage(scope, data, metaData);
diff --git a/src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/Commands/PublishEventsCommand.cs b/src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/Commands/PublishEventsCommand.cs
@@ -24,19 +24,19 @@ public class PublishEventsCommandHandler : IRequestHandler<PublishEventsCommand>
     private readonly ILogger<PublishEventsCommandHandler> _logger;
     private readonly IDateTimeProvider _dateTimeProvider;
     private readonly IRepository<OutboxEvent, Guid> _outboxEventRepository;
-    private readonly IMessageSender<AuditLogCreatedEvent> _auditLogCreatedEventSender;
+    private readonly IMessageBus _messageBus;
     private readonly DaprClient _daprClient;
 
     public PublishEventsCommandHandler(ILogger<PublishEventsCommandHandler> logger,
         IDateTimeProvider dateTimeProvider,
         IRepository<OutboxEvent, Guid> outboxEventRepository,
-        IMessageSender<AuditLogCreatedEvent> auditLogCreatedEventSender,
+        IMessageBus messageBus,
         DaprClient daprClient)
     {
         _logger = logger;
         _dateTimeProvider = dateTimeProvider;
         _outboxEventRepository = outboxEventRepository;
-        _auditLogCreatedEventSender = auditLogCreatedEventSender;
+        _messageBus = messageBus;
         _daprClient = daprClient;
     }
 
@@ -53,7 +53,7 @@ public async Task Handle(PublishEventsCommand command, CancellationToken cancell
             if (eventLog.EventType == "AUDIT_LOG_ENTRY_CREATED")
             {
                 var logEntry = JsonSerializer.Deserialize<AuditLogEntry>(eventLog.Message);
-                await _auditLogCreatedEventSender.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },
+                await _messageBus.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },
                     new MetaData
                     {
                         MessageId = eventLog.Id.ToString(),
diff --git a/src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/DTOs/AuditLogCreatedEvent.cs b/src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/DTOs/AuditLogCreatedEvent.cs
@@ -1,8 +1,9 @@
-﻿using ClassifiedAds.Services.Product.Entities;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Services.Product.Entities;
 
 namespace ClassifiedAds.Services.Product.DTOs;
 
-public class AuditLogCreatedEvent
+public class AuditLogCreatedEvent : IMessageBusEvent
 {
     public AuditLogEntry AuditLog { get; set; }
 }
diff --git a/src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/ProductModuleServiceCollectionExtensions.cs b/src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/ProductModuleServiceCollectionExtensions.cs
@@ -1,4 +1,5 @@
 ﻿using ClassifiedAds.CrossCuttingConcerns.Csv;
+using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
 using ClassifiedAds.Domain.Repositories;
 using ClassifiedAds.Infrastructure.Csv;
 using ClassifiedAds.Infrastructure.Identity;
@@ -46,7 +47,8 @@ public static IServiceCollection AddProductModule(this IServiceCollection servic
         services.AddScoped(typeof(ICsvReader<>), typeof(CsvReader<>));
         services.AddScoped(typeof(ICsvWriter<>), typeof(CsvWriter<>));
 
-        services.AddMessageBusSender<AuditLogCreatedEvent>(appSettings.MessageBroker);
+        services.AddTransient<IMessageBus, MessageBus>()
+                .AddMessageBusSender<AuditLogCreatedEvent>(appSettings.MessageBroker);
 
         return services;
     }
diff --git a/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/Commands/PublishEventsCommand.cs b/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/Commands/PublishEventsCommand.cs
@@ -23,23 +23,17 @@ public class PublishEventsCommandHandler : ICommandHandler<PublishEventsCommand>
     private readonly ILogger<PublishEventsCommandHandler> _logger;
     private readonly IDateTimeProvider _dateTimeProvider;
     private readonly IRepository<OutboxEvent, Guid> _outboxEventRepository;
-    private readonly IMessageSender<FileUploadedEvent> _fileUploadedEventSender;
-    private readonly IMessageSender<FileDeletedEvent> _fileDeletedEventSender;
-    private readonly IMessageSender<AuditLogCreatedEvent> _auditLogCreatedEventSender;
+    private readonly IMessageBus _messageBus;
 
     public PublishEventsCommandHandler(ILogger<PublishEventsCommandHandler> logger,
         IDateTimeProvider dateTimeProvider,
         IRepository<OutboxEvent, Guid> outboxEventRepository,
-        IMessageSender<FileUploadedEvent> fileUploadedEventSender,
-        IMessageSender<FileDeletedEvent> fileDeletedEventSender,
-        IMessageSender<AuditLogCreatedEvent> auditLogCreatedEventSender)
+        IMessageBus messageBus)
     {
         _logger = logger;
         _dateTimeProvider = dateTimeProvider;
         _outboxEventRepository = outboxEventRepository;
-        _fileUploadedEventSender = fileUploadedEventSender;
-        _fileDeletedEventSender = fileDeletedEventSender;
-        _auditLogCreatedEventSender = auditLogCreatedEventSender;
+        _messageBus = messageBus;
     }
 
     public async Task HandleAsync(PublishEventsCommand command, CancellationToken cancellationToken = default)
@@ -54,16 +48,16 @@ public async Task HandleAsync(PublishEventsCommand command, CancellationToken ca
         {
             if (eventLog.EventType == "FILEENTRY_CREATED")
             {
-                await _fileUploadedEventSender.SendAsync(new FileUploadedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });
+                await _messageBus.SendAsync(new FileUploadedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });
             }
             else if (eventLog.EventType == "FILEENTRY_DELETED")
             {
-                await _fileDeletedEventSender.SendAsync(new FileDeletedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });
+                await _messageBus.SendAsync(new FileDeletedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });
             }
             else if (eventLog.EventType == "AUDIT_LOG_ENTRY_CREATED")
             {
                 var logEntry = JsonSerializer.Deserialize<AuditLogEntry>(eventLog.Message);
-                await _auditLogCreatedEventSender.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },
+                await _messageBus.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },
                     new MetaData
                     {
                         MessageId = eventLog.Id.ToString(),
diff --git a/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/DTOs/AuditLogCreatedEvent.cs b/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/DTOs/AuditLogCreatedEvent.cs
@@ -1,8 +1,9 @@
-﻿using ClassifiedAds.Services.Storage.Entities;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Services.Storage.Entities;
 
 namespace ClassifiedAds.Services.Storage.DTOs;
 
-public class AuditLogCreatedEvent
+public class AuditLogCreatedEvent : IMessageBusEvent
 {
     public AuditLogEntry AuditLog { get; set; }
 }
diff --git a/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/DTOs/FileDeletedEvent.cs b/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/DTOs/FileDeletedEvent.cs
@@ -1,8 +1,9 @@
-﻿using ClassifiedAds.Services.Storage.Entities;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Services.Storage.Entities;
 
 namespace ClassifiedAds.Services.Storage.DTOs;
 
-public class FileDeletedEvent
+public class FileDeletedEvent : IMessageBusEvent
 {
     public FileEntry FileEntry { get; set; }
 }
diff --git a/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/DTOs/FileUploadedEvent.cs b/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/DTOs/FileUploadedEvent.cs
@@ -1,8 +1,9 @@
-﻿using ClassifiedAds.Services.Storage.Entities;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Services.Storage.Entities;
 
 namespace ClassifiedAds.Services.Storage.DTOs;
 
-public class FileUploadedEvent
+public class FileUploadedEvent : IMessageBusEvent
 {
     public FileEntry FileEntry { get; set; }
 }
diff --git a/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/HostedServices/MessageBusReceiver.cs b/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/HostedServices/MessageBusReceiver.cs
@@ -16,29 +16,26 @@ internal class MessageBusReceiver : BackgroundService
 
     private readonly ILogger<MessageBusReceiver> _logger;
     private readonly IConfiguration _configuration;
-    private readonly IMessageReceiver<WebhookConsumer, FileUploadedEvent> _fileUploadedEventMessageReceiver;
-    private readonly IMessageReceiver<WebhookConsumer, FileDeletedEvent> _fileDeletedEventMessageReceiver;
+    private readonly IMessageBus _messageBus;
 
     public MessageBusReceiver(ILogger<MessageBusReceiver> logger,
         IConfiguration configuration,
-        IMessageReceiver<WebhookConsumer, FileUploadedEvent> fileUploadedEventMessageReceiver,
-        IMessageReceiver<WebhookConsumer, FileDeletedEvent> fileDeletedEventMessageReceiver)
+        IMessageBus messageBus)
     {
         _logger = logger;
         _configuration = configuration;
-        _fileUploadedEventMessageReceiver = fileUploadedEventMessageReceiver;
-        _fileDeletedEventMessageReceiver = fileDeletedEventMessageReceiver;
+        _messageBus = messageBus;
     }
 
     protected override Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        _fileUploadedEventMessageReceiver?.ReceiveAsync(async (data, metaData) =>
+        _messageBus.ReceiveAsync<WebhookConsumer, FileUploadedEvent>(async (data, metaData) =>
         {
             var url = _configuration["Webhooks:FileUploadedEvent:PayloadUrl"];
             await _httpClient.PostAsJsonAsync(url, data.FileEntry);
         }, stoppingToken);
 
-        _fileDeletedEventMessageReceiver?.ReceiveAsync(async (data, metaData) =>
+        _messageBus.ReceiveAsync<WebhookConsumer, FileDeletedEvent>(async (data, metaData) =>
         {
             var url = _configuration["Webhooks:FileDeletedEvent:PayloadUrl"];
             await _httpClient.PostAsJsonAsync(url, data.FileEntry);
diff --git a/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/StorageModuleServiceCollectionExtensions.cs b/src/Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/StorageModuleServiceCollectionExtensions.cs
@@ -1,4 +1,5 @@
-﻿using ClassifiedAds.Domain.Repositories;
+﻿using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.Domain.Repositories;
 using ClassifiedAds.Infrastructure.Identity;
 using ClassifiedAds.Services.Storage.Authorization;
 using ClassifiedAds.Services.Storage.ConfigurationOptions;
@@ -38,7 +39,8 @@ public static IServiceCollection AddStorageModule(this IServiceCollection servic
 
         services.AddStorageManager(appSettings.Storage);
 
-        services.AddMessageBusSender<FileUploadedEvent>(appSettings.MessageBroker)
+        services.AddTransient<IMessageBus, MessageBus>()
+                .AddMessageBusSender<FileUploadedEvent>(appSettings.MessageBroker)
                 .AddMessageBusSender<FileDeletedEvent>(appSettings.MessageBroker)
                 .AddMessageBusSender<AuditLogCreatedEvent>(appSettings.MessageBroker)
                 .AddMessageBusReceiver<WebhookConsumer, FileUploadedEvent>(appSettings.MessageBroker)

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
++
 +public interface IMessageBusEvent
 +{
 +}
Original file line number	Diff line number	Diff line change
`@@ -1,8 +1,9 @@`
`1`		`-using ClassifiedAds.Services.AuditLog.Entities;`
	`1`	`+using ClassifiedAds.Domain.Infrastructure.MessageBrokers;`
	`2`	`+using ClassifiedAds.Services.AuditLog.Entities;`
`2`	`3`
`3`	`4`	`namespace ClassifiedAds.Services.AuditLog.DTOs;`
`4`	`5`
`5`		`-public class AuditLogCreatedEvent`
	`6`	`+public class AuditLogCreatedEvent : IMessageBusEvent`
`6`	`7`	`{`
`7`	`8`	`public AuditLogEntry AuditLog { get; set; }`
`8`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,20 +14,20 @@ internal class MessageBusReceiver : BackgroundService`
`14`	`14`	`{`
`15`	`15`	`private readonly ILogger<MessageBusReceiver> _logger;`
`16`	`16`	`private readonly IServiceProvider _serviceProvider;`
`17`		`- private readonly IMessageReceiver<AuditLogAggregationConsumer, AuditLogCreatedEvent> _auditLogCreatedEventReceiver;`
	`17`	`+ private readonly IMessageBus _messageBus;`
`18`	`18`
`19`	`19`	`public MessageBusReceiver(ILogger<MessageBusReceiver> logger,`
`20`	`20`	`IServiceProvider serviceProvider,`
`21`		`- IMessageReceiver<AuditLogAggregationConsumer, AuditLogCreatedEvent> auditLogCreatedEventReceiver)`
	`21`	`+ IMessageBus messageBus)`
`22`	`22`	`{`
`23`	`23`	`_logger = logger;`
`24`	`24`	`_serviceProvider = serviceProvider;`
`25`		`- _auditLogCreatedEventReceiver = auditLogCreatedEventReceiver;`
	`25`	`+ _messageBus = messageBus;`
`26`	`26`	`}`
`27`	`27`
`28`	`28`	`protected override Task ExecuteAsync(CancellationToken stoppingToken)`
`29`	`29`	`{`
`30`		`- _auditLogCreatedEventReceiver?.ReceiveAsync(async (data, metaData) =>`
	`30`	`+ _messageBus.ReceiveAsync<AuditLogAggregationConsumer, AuditLogCreatedEvent>(async (data, metaData) =>`
`31`	`31`	`{`
`32`	`32`	`using var scope = _serviceProvider.CreateScope();`
`33`	`33`	`await ProcessMessage(scope, data, metaData);`
Original file line number	Diff line number	Diff line change
`@@ -1,8 +1,9 @@`
`1`		`-using ClassifiedAds.Services.Product.Entities;`
	`1`	`+using ClassifiedAds.Domain.Infrastructure.MessageBrokers;`
	`2`	`+using ClassifiedAds.Services.Product.Entities;`
`2`	`3`
`3`	`4`	`namespace ClassifiedAds.Services.Product.DTOs;`
`4`	`5`
`5`		`-public class AuditLogCreatedEvent`
	`6`	`+public class AuditLogCreatedEvent : IMessageBusEvent`
`6`	`7`	`{`
`7`	`8`	`public AuditLogEntry AuditLog { get; set; }`
`8`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -23,23 +23,17 @@ public class PublishEventsCommandHandler : ICommandHandler<PublishEventsCommand>`
`23`	`23`	`private readonly ILogger<PublishEventsCommandHandler> _logger;`
`24`	`24`	`private readonly IDateTimeProvider _dateTimeProvider;`
`25`	`25`	`private readonly IRepository<OutboxEvent, Guid> _outboxEventRepository;`
`26`		`- private readonly IMessageSender<FileUploadedEvent> _fileUploadedEventSender;`
`27`		`- private readonly IMessageSender<FileDeletedEvent> _fileDeletedEventSender;`
`28`		`- private readonly IMessageSender<AuditLogCreatedEvent> _auditLogCreatedEventSender;`
	`26`	`+ private readonly IMessageBus _messageBus;`
`29`	`27`
`30`	`28`	`public PublishEventsCommandHandler(ILogger<PublishEventsCommandHandler> logger,`
`31`	`29`	`IDateTimeProvider dateTimeProvider,`
`32`	`30`	`IRepository<OutboxEvent, Guid> outboxEventRepository,`
`33`		`- IMessageSender<FileUploadedEvent> fileUploadedEventSender,`
`34`		`- IMessageSender<FileDeletedEvent> fileDeletedEventSender,`
`35`		`- IMessageSender<AuditLogCreatedEvent> auditLogCreatedEventSender)`
	`31`	`+ IMessageBus messageBus)`
`36`	`32`	`{`
`37`	`33`	`_logger = logger;`
`38`	`34`	`_dateTimeProvider = dateTimeProvider;`
`39`	`35`	`_outboxEventRepository = outboxEventRepository;`
`40`		`- _fileUploadedEventSender = fileUploadedEventSender;`
`41`		`- _fileDeletedEventSender = fileDeletedEventSender;`
`42`		`- _auditLogCreatedEventSender = auditLogCreatedEventSender;`
	`36`	`+ _messageBus = messageBus;`
`43`	`37`	`}`
`44`	`38`
`45`	`39`	`public async Task HandleAsync(PublishEventsCommand command, CancellationToken cancellationToken = default)`
`@@ -54,16 +48,16 @@ public async Task HandleAsync(PublishEventsCommand command, CancellationToken ca`
`54`	`48`	`{`
`55`	`49`	`if (eventLog.EventType == "FILEENTRY_CREATED")`
`56`	`50`	`{`
`57`		`- await _fileUploadedEventSender.SendAsync(new FileUploadedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });`
	`51`	`+ await _messageBus.SendAsync(new FileUploadedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });`
`58`	`52`	`}`
`59`	`53`	`else if (eventLog.EventType == "FILEENTRY_DELETED")`
`60`	`54`	`{`
`61`		`- await _fileDeletedEventSender.SendAsync(new FileDeletedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });`
	`55`	`+ await _messageBus.SendAsync(new FileDeletedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(eventLog.Message) });`
`62`	`56`	`}`
`63`	`57`	`else if (eventLog.EventType == "AUDIT_LOG_ENTRY_CREATED")`
`64`	`58`	`{`
`65`	`59`	`var logEntry = JsonSerializer.Deserialize<AuditLogEntry>(eventLog.Message);`
`66`		`- await _auditLogCreatedEventSender.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },`
	`60`	`+ await _messageBus.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },`
`67`	`61`	`new MetaData`
`68`	`62`	`{`
`69`	`63`	`MessageId = eventLog.Id.ToString(),`