Skip to content

Commit 9dbc98c

Browse files
phongnguyendphongnguyend
committed
Message Bus Consumers + OutBox Event Publishers
1 parent 574549a commit 9dbc98c

File tree

41 files changed

+1130
-388
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

41 files changed

+1130
-388
lines changed

‎src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageBus.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,11 @@ Task SendAsync<T>(T message, MetaData metaData = null, CancellationToken cancell
1111

1212
Task ReceiveAsync<TConsumer, T>(Func<T, MetaData, Task> action, CancellationToken cancellationToken = default)
1313
where T : IMessageBusMessage;
14+
15+
Task ReceiveAsync<TConsumer, T>(CancellationToken cancellationToken = default)
16+
where T : IMessageBusMessage;
17+
18+
Task SendAsync(PublishingOutBoxEvent outbox, CancellationToken cancellationToken = default);
1419
}
1520

1621
public interface IMessageBusMessage

‎src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageBusConsumer.cs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
using System.Threading;
2+
using System.Threading.Tasks;
3+
4+
namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
5+
6+
public interface IMessageBusConsumer<TConsumer, T>
7+
{
8+
Task HandleAsync(T data, MetaData metaData, CancellationToken cancellationToken = default);
9+
}

‎src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IOutBoxEventPublisher.cs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
using System.Threading;
2+
using System.Threading.Tasks;
3+
4+
namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
5+
6+
public interface IOutBoxEventPublisher
7+
{
8+
static abstract string[] CanHandleEventTypes();
9+
10+
static abstract string CanHandleEventSource();
11+
12+
Task HandleAsync(PublishingOutBoxEvent outbox, CancellationToken cancellationToken = default);
13+
}
14+
15+
public class PublishingOutBoxEvent
16+
{
17+
public string Id { get; set; }
18+
19+
public string EventType { get; set; }
20+
21+
public string EventSource { get; set; }
22+
23+
public string Payload { get; set; }
24+
}

‎src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MessageBus.cs‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
using Microsoft.Extensions.DependencyInjection;
22
using System;
3+
using System.Collections.Generic;
4+
using System.Globalization;
5+
using System.Linq;
6+
using System.Reflection;
37
using System.Threading;
48
using System.Threading.Tasks;
59

@@ -8,6 +12,51 @@ namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
812
public class MessageBus : IMessageBus
913
{
1014
private readonly IServiceProvider _serviceProvider;
15+
private static List<Type> _consumers = new List<Type>();
16+
private static Dictionary<string, List<Type>> _outboxEventHandlers = new();
17+
18+
internal static void AddConsumers(Assembly assembly, IServiceCollection services)
19+
{
20+
var types = assembly.GetTypes()
21+
.Where(x => x.GetInterfaces().Any(y => y.IsGenericType && y.GetGenericTypeDefinition() == typeof(IMessageBusConsumer<,>)))
22+
.ToList();
23+
24+
foreach (var type in types)
25+
{
26+
services.AddTransient(type);
27+
}
28+
29+
_consumers.AddRange(types);
30+
}
31+
32+
internal static void AddOutboxEventPublishers(Assembly assembly, IServiceCollection services)
33+
{
34+
var types = assembly.GetTypes()
35+
.Where(x => x.GetInterfaces().Any(y => y == typeof(IOutBoxEventPublisher)))
36+
.ToList();
37+
38+
foreach (var type in types)
39+
{
40+
services.AddTransient(type);
41+
}
42+
43+
foreach (var item in types)
44+
{
45+
var canHandlerEventTypes = (string[])item.InvokeMember(nameof(IOutBoxEventPublisher.CanHandleEventTypes), BindingFlags.InvokeMethod, null, null, null, CultureInfo.CurrentCulture);
46+
var eventSource = (string)item.InvokeMember(nameof(IOutBoxEventPublisher.CanHandleEventSource), BindingFlags.InvokeMethod, null, null, null, CultureInfo.CurrentCulture);
47+
48+
foreach (var eventType in canHandlerEventTypes)
49+
{
50+
var key = eventSource + ":" + eventType;
51+
if (!_outboxEventHandlers.ContainsKey(key))
52+
{
53+
_outboxEventHandlers[key] = new List<Type>();
54+
}
55+
56+
_outboxEventHandlers[key].Add(item);
57+
}
58+
}
59+
}
1160

1261
public MessageBus(IServiceProvider serviceProvider)
1362
{
@@ -25,4 +74,64 @@ public async Task ReceiveAsync<TConsumer, T>(Func<T, MetaData, Task> action, Can
2574
{
2675
await _serviceProvider.GetRequiredService<IMessageReceiver<TConsumer, T>>().ReceiveAsync(action, cancellationToken);
2776
}
77+
78+
public async Task ReceiveAsync<TConsumer, T>(CancellationToken cancellationToken = default)
79+
where T : IMessageBusMessage
80+
{
81+
await _serviceProvider.GetRequiredService<IMessageReceiver<TConsumer, T>>().ReceiveAsync(async (data, metaData) =>
82+
{
83+
using var scope = _serviceProvider.CreateScope();
84+
foreach (Type handlerType in _consumers)
85+
{
86+
bool canHandleEvent = handlerType.GetInterfaces()
87+
.Any(x => x.IsGenericType
88+
&& x.GetGenericTypeDefinition() == typeof(IMessageBusConsumer<,>)
89+
&& x.GenericTypeArguments[0] == typeof(TConsumer) && x.GenericTypeArguments[1] == typeof(T));
90+
91+
if (canHandleEvent)
92+
{
93+
dynamic handler = scope.ServiceProvider.GetService(handlerType);
94+
await handler.HandleAsync((dynamic)data, metaData, cancellationToken);
95+
}
96+
}
97+
}, cancellationToken);
98+
}
99+
100+
public async Task SendAsync(PublishingOutBoxEvent outbox, CancellationToken cancellationToken = default)
101+
{
102+
var key = outbox.EventSource + ":" + outbox.EventType;
103+
var handlerTypes = _outboxEventHandlers.ContainsKey(key) ? _outboxEventHandlers[key] : null;
104+
105+
if (handlerTypes == null)
106+
{
107+
// TODO: Take Note
108+
return;
109+
}
110+
111+
foreach (var type in handlerTypes)
112+
{
113+
dynamic handler = _serviceProvider.GetService(type);
114+
await handler.HandleAsync(outbox, cancellationToken);
115+
}
116+
}
28117
}
118+
119+
public static class MessageBusExtentions
120+
{
121+
public static void AddMessageBusConsumers(this IServiceCollection services, Assembly assembly)
122+
{
123+
MessageBus.AddConsumers(assembly, services);
124+
}
125+
126+
public static void AddOutboxEventPublishers(this IServiceCollection services, Assembly assembly)
127+
{
128+
MessageBus.AddOutboxEventPublishers(assembly, services);
129+
}
130+
131+
public static void AddMessageBus(this IServiceCollection services, Assembly assembly)
132+
{
133+
services.AddTransient<IMessageBus, MessageBus>();
134+
services.AddMessageBusConsumers(assembly);
135+
services.AddOutboxEventPublishers(assembly);
136+
}
137+
}

‎src/Microservices/Common/ClassifiedAds.Infrastructure/HostedServices/MessageBusConsumerBackgroundService.cs‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
2+
using Microsoft.Extensions.Hosting;
3+
using Microsoft.Extensions.Logging;
4+
using System.Threading;
5+
using System.Threading.Tasks;
6+
7+
namespace ClassifiedAds.Infrastructure.HostedServices;
8+
9+
public sealed class MessageBusConsumerBackgroundService<TConsumer, T> : BackgroundService
10+
where T : IMessageBusEvent
11+
{
12+
private readonly ILogger<MessageBusConsumerBackgroundService<TConsumer, T>> _logger;
13+
private readonly IMessageBus _messageBus;
14+
15+
public MessageBusConsumerBackgroundService(ILogger<MessageBusConsumerBackgroundService<TConsumer, T>> logger,
16+
IMessageBus messageBus)
17+
{
18+
_logger = logger;
19+
_messageBus = messageBus;
20+
}
21+
22+
protected override Task ExecuteAsync(CancellationToken stoppingToken)
23+
{
24+
_messageBus.ReceiveAsync<TConsumer, T>(stoppingToken);
25+
return Task.CompletedTask;
26+
}
27+
}

‎src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/AuditLogModuleServiceCollectionExtensions.cs‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
22
using ClassifiedAds.Domain.Repositories;
3+
using ClassifiedAds.Infrastructure.HostedServices;
34
using ClassifiedAds.Services.AuditLog.Authorization;
45
using ClassifiedAds.Services.AuditLog.ConfigurationOptions;
56
using ClassifiedAds.Services.AuditLog.DTOs;
67
using ClassifiedAds.Services.AuditLog.Entities;
7-
using ClassifiedAds.Services.AuditLog.HostedServices;
8+
using ClassifiedAds.Services.AuditLog.MessageBusConsumers;
89
using ClassifiedAds.Services.AuditLog.Repositories;
910
using Microsoft.AspNetCore.Builder;
1011
using Microsoft.EntityFrameworkCore;
@@ -46,7 +47,10 @@ public static void MigrateAuditLogDb(this IApplicationBuilder app)
4647

4748
public static IServiceCollection AddHostedServicesAuditLogModule(this IServiceCollection services)
4849
{
49-
services.AddHostedService<MessageBusReceiver>();
50+
services.AddMessageBusConsumers(Assembly.GetExecutingAssembly());
51+
services.AddOutboxEventPublishers(Assembly.GetExecutingAssembly());
52+
53+
services.AddHostedService<MessageBusConsumerBackgroundService<AuditLogAggregationConsumer, AuditLogCreatedEvent>>();
5054

5155
return services;
5256
}

‎src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/HostedServices/MessageBusReceiver.cs‎ renamed to ‎src/Microservices/Services.AuditLog/ClassifiedAds.Services.AuditLog/MessageBusConsumers/AuditLogAggregationConsumer.cs‎

Lines changed: 8 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -5,41 +5,26 @@
55
using ClassifiedAds.Services.AuditLog.Entities;
66
using Microsoft.EntityFrameworkCore;
77
using Microsoft.Extensions.DependencyInjection;
8-
using Microsoft.Extensions.Hosting;
98
using Microsoft.Extensions.Logging;
109

11-
namespace ClassifiedAds.Services.AuditLog.HostedServices;
10+
namespace ClassifiedAds.Services.AuditLog.MessageBusConsumers;
1211

13-
internal class MessageBusReceiver : BackgroundService
12+
public sealed class AuditLogAggregationConsumer : IMessageBusConsumer<AuditLogAggregationConsumer, AuditLogCreatedEvent>
1413
{
15-
private readonly ILogger<MessageBusReceiver> _logger;
14+
private readonly ILogger<AuditLogAggregationConsumer> _logger;
1615
private readonly IServiceProvider _serviceProvider;
17-
private readonly IMessageBus _messageBus;
1816

19-
public MessageBusReceiver(ILogger<MessageBusReceiver> logger,
20-
IServiceProvider serviceProvider,
21-
IMessageBus messageBus)
17+
public AuditLogAggregationConsumer(ILogger<AuditLogAggregationConsumer> logger,
18+
IServiceProvider serviceProvider)
2219
{
2320
_logger = logger;
2421
_serviceProvider = serviceProvider;
25-
_messageBus = messageBus;
2622
}
2723

28-
protected override Task ExecuteAsync(CancellationToken stoppingToken)
24+
public async Task HandleAsync(AuditLogCreatedEvent data, MetaData metaData, CancellationToken cancellationToken = default)
2925
{
30-
_messageBus.ReceiveAsync<AuditLogAggregationConsumer, AuditLogCreatedEvent>(async (data, metaData) =>
31-
{
32-
using var scope = _serviceProvider.CreateScope();
33-
await ProcessMessage(scope, data, metaData);
34-
}, stoppingToken);
35-
36-
return Task.CompletedTask;
37-
}
38-
39-
private async Task ProcessMessage(IServiceScope scope, AuditLogCreatedEvent data, MetaData metaData)
40-
{
41-
var dispatcher = scope.ServiceProvider.GetRequiredService<Dispatcher>();
42-
var idempotentRequestRepository = scope.ServiceProvider.GetRequiredService<IRepository<IdempotentRequest, Guid>>();
26+
var dispatcher = _serviceProvider.GetRequiredService<Dispatcher>();
27+
var idempotentRequestRepository = _serviceProvider.GetRequiredService<IRepository<IdempotentRequest, Guid>>();
4328

4429
var requestType = "ADD_AUDIT_LOG_ENTRY";
4530

@@ -70,7 +55,3 @@ await idempotentRequestRepository.AddAsync(new IdempotentRequest
7055
await uow.CommitTransactionAsync();
7156
}
7257
}
73-
74-
public sealed class AuditLogAggregationConsumer
75-
{
76-
}

‎src/Microservices/Services.Product/ClassifiedAds.Services.Product.Api/Commands/PublishEventsCommand.cs‎

Lines changed: 8 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,11 @@
11
using ClassifiedAds.CrossCuttingConcerns.DateTimes;
22
using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
33
using ClassifiedAds.Domain.Repositories;
4-
using ClassifiedAds.Services.Product.Constants;
5-
using ClassifiedAds.Services.Product.DTOs;
64
using ClassifiedAds.Services.Product.Entities;
7-
using Dapr.Client;
85
using MediatR;
96
using Microsoft.Extensions.Logging;
107
using System;
118
using System.Linq;
12-
using System.Text.Json;
139
using System.Threading;
1410
using System.Threading.Tasks;
1511

@@ -26,19 +22,16 @@ public class PublishEventsCommandHandler : IRequestHandler<PublishEventsCommand>
2622
private readonly IDateTimeProvider _dateTimeProvider;
2723
private readonly IRepository<OutboxEvent, Guid> _outboxEventRepository;
2824
private readonly IMessageBus _messageBus;
29-
private readonly DaprClient _daprClient;
3025

3126
public PublishEventsCommandHandler(ILogger<PublishEventsCommandHandler> logger,
3227
IDateTimeProvider dateTimeProvider,
3328
IRepository<OutboxEvent, Guid> outboxEventRepository,
34-
IMessageBus messageBus,
35-
DaprClient daprClient)
29+
IMessageBus messageBus)
3630
{
3731
_logger = logger;
3832
_dateTimeProvider = dateTimeProvider;
3933
_outboxEventRepository = outboxEventRepository;
4034
_messageBus = messageBus;
41-
_daprClient = daprClient;
4235
}
4336

4437
public async Task Handle(PublishEventsCommand command, CancellationToken cancellationToken = default)
@@ -51,24 +44,15 @@ public async Task Handle(PublishEventsCommand command, CancellationToken cancell
5144

5245
foreach (var eventLog in events)
5346
{
54-
if (eventLog.EventType == EventTypeConstants.AuditLogEntryCreated)
47+
var outbox = new PublishingOutBoxEvent
5548
{
56-
var logEntry = JsonSerializer.Deserialize<AuditLogEntry>(eventLog.Message);
57-
await _messageBus.SendAsync(new AuditLogCreatedEvent { AuditLog = logEntry },
58-
new MetaData
59-
{
60-
MessageId = eventLog.Id.ToString(),
61-
});
49+
Id = eventLog.Id.ToString(),
50+
EventType = eventLog.EventType,
51+
EventSource = typeof(PublishEventsCommand).Assembly.GetName().Name,
52+
Payload = eventLog.Message,
53+
};
6254

63-
if (!string.IsNullOrWhiteSpace(Environment.GetEnvironmentVariable("DAPR_HTTP_PORT")))
64-
{
65-
await _daprClient.PublishEventAsync("pubsub", "AuditLogCreatedEvent", new AuditLogCreatedEvent { AuditLog = logEntry });
66-
}
67-
}
68-
else
69-
{
70-
// TODO: Take Note
71-
}
55+
await _messageBus.SendAsync(outbox, cancellationToken);
7256

7357
eventLog.Published = true;
7458
eventLog.UpdatedDateTime = _dateTimeProvider.OffsetNow;

0 commit comments

Comments
 (0)