From 5857e2738d422abb51a4c14ee495ef9fa5d4cfc6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Timo=20H=C3=BCgel?= Date: Wed, 26 Aug 2026 09:47:45 +0200 Subject: [PATCH 1/3] feat: introduce partial classes for rabbitmq --- .../RabbitMQ/EventBusRabbitMq.Consumer.cs | 133 ++++++++++++++ .../RabbitMQ/EventBusRabbitMq.Publisher.cs | 55 ++++++ .../EventBus/RabbitMQ/EventBusRabbitMq.cs | 170 +----------------- 3 files changed, 190 insertions(+), 168 deletions(-) create mode 100644 BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Consumer.cs create mode 100644 BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Publisher.cs diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Consumer.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Consumer.cs new file mode 100644 index 0000000000..caef784981 --- /dev/null +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Consumer.cs @@ -0,0 +1,133 @@ +using System.Diagnostics; +using System.Text; +using System.Text.Json; +using Backbone.BuildingBlocks.Application.Abstractions.Infrastructure.EventBus; +using Backbone.BuildingBlocks.Domain.Events; +using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; +using Backbone.Tooling.Extensions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +namespace Backbone.BuildingBlocks.Infrastructure.EventBus.RabbitMQ; + +public partial class EventBusRabbitMq +{ + public async Task Subscribe() where TEvent : DomainEvent where THandler : IDomainEventHandler + { + var queueName = GetQueueName(); + + await CreateQueue(queueName); + + var consumer = await CreateConsumer(); + + _subscriptionManager.AddSubscription(consumer, queueName); + } + + private async Task CreateQueue(string queueName) where TEvent : DomainEvent + { + var eventName = typeof(TEvent).GetEventName(); + + var channel = await _channelPool.Get(); + + await channel.QueueDeclareAsync(queueName, + durable: true, + exclusive: false, + autoDelete: false, + arguments: new Dictionary + { + { "x-queue-type", "quorum" }, + { "x-delivery-limit", HANDLER_RETRY_COUNT }, + { "x-dead-letter-exchange", _deadLetterExchangeName }, + { "x-dead-letter-routing-key", $"dead.routing.{eventName}" } + } + ); + + await channel.QueueBindAsync(queueName, _exchangeName, eventName); + + _logger.LogTrace("Successfully bound queue '{QueueName}' to event '{EventName}'.", queueName, eventName); + + _channelPool.Return(channel); + } + + private async Task CreateConsumer() where TEvent : DomainEvent where THandler : IDomainEventHandler + { + var channel = await _connection.CreateChannelAsync(); + var consumer = new AsyncEventingBasicConsumer(channel); + + consumer.ReceivedAsync += async (_, eventArgs) => + { + var eventName = eventArgs.RoutingKey; + + var message = Encoding.UTF8.GetString(eventArgs.Body.ToArray()); + + try + { + var correlationId = eventArgs.BasicProperties.CorrelationId; + correlationId = correlationId.IsNullOrEmpty() ? CustomLogContext.GenerateCorrelationId() : correlationId; + + using (CustomLogContext.SetCorrelationId(correlationId)) + { + await ProcessEvent(message); + + await channel.BasicAckAsync(eventArgs.DeliveryTag, false); + } + } + catch (Exception ex) + { + await channel.BasicRejectAsync(eventArgs.DeliveryTag, true); + _metrics.IncrementNumberOfProcessingErrors(GetQueueName()); + + _logger.ErrorWhileProcessingDomainEvent(eventName, ex); + } + }; + + return consumer; + } + + private async Task ProcessEvent(string message) where TEvent : DomainEvent where THandler : IDomainEventHandler + { + var eventType = typeof(TEvent); + var eventName = eventType.GetEventName(); + + _logger.LogDebug("Processing RabbitMQ event: '{EventName}'", eventName); + + var domainEvent = JsonSerializer.Deserialize(message); + + var handlerType = typeof(THandler); + + await using var scope = _serviceProvider.CreateAsyncScope(); + + if (scope.ServiceProvider.GetService(handlerType) is not IDomainEventHandler handler) + throw new Exception("Domain event handler could not be resolved from dependency container or it does not implement IDomainEventHandler."); + + var concreteType = typeof(IDomainEventHandler<>).MakeGenericType(eventType); + + var startedAt = Stopwatch.GetTimestamp(); + await (Task)concreteType.GetMethod("Handle")!.Invoke(handler, [domainEvent])!; + _metrics.TrackEventProcessingDuration(startedAt, GetQueueName()); + + _metrics.IncrementNumberOfHandledEvents(GetQueueName()); + } + + public async Task StartConsuming(CancellationToken cancellationToken) + { + foreach (var subscription in _subscriptionManager) + { + await subscription.Consumer.Channel.BasicConsumeAsync(subscription.QueueName, autoAck: false, subscription.Consumer, cancellationToken); + } + } + + public async Task StopConsuming(CancellationToken cancellationToken) + { + foreach (var consumerData in _subscriptionManager) + { + var channel = consumerData.Consumer.Channel; + foreach (var tag in consumerData.Consumer.ConsumerTags) + { + await channel.BasicCancelAsync(tag, cancellationToken: cancellationToken); + } + } + } +} diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Publisher.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Publisher.cs new file mode 100644 index 0000000000..4a89b25b16 --- /dev/null +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.Publisher.cs @@ -0,0 +1,55 @@ +using System.Diagnostics; +using System.Text; +using System.Text.Json; +using Backbone.BuildingBlocks.Domain.Events; +using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; +using Microsoft.Extensions.Logging; +using RabbitMQ.Client; + +namespace Backbone.BuildingBlocks.Infrastructure.EventBus.RabbitMQ; + +public partial class EventBusRabbitMq +{ + public async Task Publish(DomainEvent @event) + { + var eventName = @event.GetEventName(); + + _logger.LogInformation("Creating RabbitMQ channel to publish a '{EventName}'.", eventName); + + var message = JsonSerializer.Serialize(@event, @event.GetType()); + + var body = Encoding.UTF8.GetBytes(message); + + _metrics.TrackHandledMessageSize(body.Length); + + await _publishRetryPolicy.ExecuteAsync(async () => + { + _logger.LogDebug("Publishing a '{EventName}' event to RabbitMQ.", eventName); + + var channel = await _channelPool.Get(); + var properties = new BasicProperties + { + DeliveryMode = DeliveryModes.Persistent, + MessageId = @event.DomainEventId, + CorrelationId = CustomLogContext.GetCorrelationId() + }; + + try + { + var startedAt = Stopwatch.GetTimestamp(); + await channel.BasicPublishAsync(_exchangeName, eventName, mandatory: false, properties, body); + _logger.PublishedDomainEvent(); + + _metrics.TrackEventPublishingDuration(startedAt); + _metrics.IncrementNumberOfPublishedEvents(eventName); + } + catch (Exception) + { + _metrics.IncrementNumberOfPublishingErrors(eventName); + throw; + } + + _channelPool.Return(channel); + }); + } +} diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.cs index b02b02caea..aca59ab97d 100644 --- a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.cs +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/RabbitMQ/EventBusRabbitMq.cs @@ -1,22 +1,16 @@ -using System.Diagnostics; -using System.Net.Sockets; -using System.Text; -using System.Text.Json; +using System.Net.Sockets; using Backbone.BuildingBlocks.Application.Abstractions.Infrastructure.EventBus; using Backbone.BuildingBlocks.Domain.Events; -using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; using Backbone.Tooling.Extensions; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Polly; using Polly.Retry; using RabbitMQ.Client; -using RabbitMQ.Client.Events; using RabbitMQ.Client.Exceptions; namespace Backbone.BuildingBlocks.Infrastructure.EventBus.RabbitMQ; -public class EventBusRabbitMq : IEventBus, IDisposable +public partial class EventBusRabbitMq : IEventBus, IDisposable { private const int PUBLISH_RETRY_COUNT = 6; private const int CONNECTION_RETRY_COUNT = 6; @@ -146,166 +140,6 @@ await channel.QueueDeclareAsync(_deadLetterQueueName, _channelPool.Return(channel); } - public async Task Subscribe() where TEvent : DomainEvent where THandler : IDomainEventHandler - { - var queueName = GetQueueName(); - - await CreateQueue(queueName); - - var consumer = await CreateConsumer(); - - _subscriptionManager.AddSubscription(consumer, queueName); - } - - private async Task CreateQueue(string queueName) where TEvent : DomainEvent - { - var eventName = typeof(TEvent).GetEventName(); - - var channel = await _channelPool.Get(); - - await channel.QueueDeclareAsync(queueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: new Dictionary - { - { "x-queue-type", "quorum" }, - { "x-delivery-limit", HANDLER_RETRY_COUNT }, - { "x-dead-letter-exchange", _deadLetterExchangeName }, - { "x-dead-letter-routing-key", $"dead.routing.{eventName}" } - } - ); - - await channel.QueueBindAsync(queueName, _exchangeName, eventName); - - _logger.LogTrace("Successfully bound queue '{QueueName}' to event '{EventName}'.", queueName, eventName); - - _channelPool.Return(channel); - } - - private async Task CreateConsumer() where TEvent : DomainEvent where THandler : IDomainEventHandler - { - var channel = await _connection.CreateChannelAsync(); - var consumer = new AsyncEventingBasicConsumer(channel); - - consumer.ReceivedAsync += async (_, eventArgs) => - { - var eventName = eventArgs.RoutingKey; - - var message = Encoding.UTF8.GetString(eventArgs.Body.ToArray()); - - try - { - var correlationId = eventArgs.BasicProperties.CorrelationId; - correlationId = correlationId.IsNullOrEmpty() ? CustomLogContext.GenerateCorrelationId() : correlationId; - - using (CustomLogContext.SetCorrelationId(correlationId)) - { - await ProcessEvent(message); - - await channel.BasicAckAsync(eventArgs.DeliveryTag, false); - } - } - catch (Exception ex) - { - await channel.BasicRejectAsync(eventArgs.DeliveryTag, true); - _metrics.IncrementNumberOfProcessingErrors(GetQueueName()); - - _logger.ErrorWhileProcessingDomainEvent(eventName, ex); - } - }; - - return consumer; - } - - private async Task ProcessEvent(string message) where TEvent : DomainEvent where THandler : IDomainEventHandler - { - var eventType = typeof(TEvent); - var eventName = eventType.GetEventName(); - - _logger.LogDebug("Processing RabbitMQ event: '{EventName}'", eventName); - - var domainEvent = JsonSerializer.Deserialize(message); - - var handlerType = typeof(THandler); - - await using var scope = _serviceProvider.CreateAsyncScope(); - - if (scope.ServiceProvider.GetService(handlerType) is not IDomainEventHandler handler) - throw new Exception("Domain event handler could not be resolved from dependency container or it does not implement IDomainEventHandler."); - - var concreteType = typeof(IDomainEventHandler<>).MakeGenericType(eventType); - - var startedAt = Stopwatch.GetTimestamp(); - await (Task)concreteType.GetMethod("Handle")!.Invoke(handler, [domainEvent])!; - _metrics.TrackEventProcessingDuration(startedAt, GetQueueName()); - - _metrics.IncrementNumberOfHandledEvents(GetQueueName()); - } - - public async Task Publish(DomainEvent @event) - { - var eventName = @event.GetEventName(); - - _logger.LogInformation("Creating RabbitMQ channel to publish a '{EventName}'.", eventName); - - var message = JsonSerializer.Serialize(@event, @event.GetType()); - - var body = Encoding.UTF8.GetBytes(message); - - _metrics.TrackHandledMessageSize(body.Length); - - await _publishRetryPolicy.ExecuteAsync(async () => - { - _logger.LogDebug("Publishing a '{EventName}' event to RabbitMQ.", eventName); - - var channel = await _channelPool.Get(); - var properties = new BasicProperties - { - DeliveryMode = DeliveryModes.Persistent, - MessageId = @event.DomainEventId, - CorrelationId = CustomLogContext.GetCorrelationId() - }; - - try - { - var startedAt = Stopwatch.GetTimestamp(); - await channel.BasicPublishAsync(_exchangeName, eventName, mandatory: false, properties, body); - _logger.PublishedDomainEvent(); - - _metrics.TrackEventPublishingDuration(startedAt); - _metrics.IncrementNumberOfPublishedEvents(eventName); - } - catch (Exception) - { - _metrics.IncrementNumberOfPublishingErrors(eventName); - throw; - } - - _channelPool.Return(channel); - }); - } - - public async Task StartConsuming(CancellationToken cancellationToken) - { - foreach (var subscription in _subscriptionManager) - { - await subscription.Consumer.Channel.BasicConsumeAsync(subscription.QueueName, autoAck: false, subscription.Consumer, cancellationToken); - } - } - - public async Task StopConsuming(CancellationToken cancellationToken) - { - foreach (var consumerData in _subscriptionManager) - { - var channel = consumerData.Consumer.Channel; - foreach (var tag in consumerData.Consumer.ConsumerTags) - { - await channel.BasicCancelAsync(tag, cancellationToken: cancellationToken); - } - } - } - public void Dispose() { _channelPool.Dispose(); From 121a366b485345c1e8ab9f8f8cb2e84dacd786fe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Timo=20H=C3=BCgel?= Date: Wed, 26 Aug 2026 09:48:52 +0200 Subject: [PATCH 2/3] feat: introduce partial classes for AzureServiceBus --- .../EventBusAzureServiceBus.Consumer.cs | 154 +++++++++++++++ .../EventBusAzureServiceBus.Publisher.cs | 47 +++++ .../EventBusAzureServiceBus.cs | 181 +----------------- 3 files changed, 202 insertions(+), 180 deletions(-) create mode 100644 BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Consumer.cs create mode 100644 BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Publisher.cs diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Consumer.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Consumer.cs new file mode 100644 index 0000000000..a6fe12fcd3 --- /dev/null +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Consumer.cs @@ -0,0 +1,154 @@ +using System.Diagnostics; +using System.Text.Json; +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; +using Backbone.BuildingBlocks.Application.Abstractions.Infrastructure.EventBus; +using Backbone.BuildingBlocks.Domain.Events; +using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; +using Backbone.Tooling.Extensions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace Backbone.BuildingBlocks.Infrastructure.EventBus.AzureServiceBus; + +public partial class EventBusAzureServiceBus +{ + public async Task Subscribe() + where T : DomainEvent + where TH : IDomainEventHandler + { + var eventName = typeof(T).GetEventName(); + var subscriptionName = GetSubscriptionName(); + + await EnsureSubscriptionExists(subscriptionName); + + await RegisterSubscriptionForEvent(subscriptionName, eventName); + + var processor = _client.CreateProcessor(TOPIC_NAME, subscriptionName, _options); + + processor.ProcessMessageAsync += + async args => + { + var messageData = args.Message.Body.ToString(); + var correlationId = args.Message.CorrelationId; + + correlationId = correlationId.IsNullOrEmpty() ? CustomLogContext.GenerateCorrelationId() : correlationId; + + using (CustomLogContext.SetCorrelationId(correlationId)) + { + var processedSuccessfully = await ProcessEvent(messageData); + + if (processedSuccessfully) + await args.CompleteMessageAsync(args.Message); + else + { + await args.AbandonMessageAsync(args.Message); + _metrics.IncrementNumberOfProcessingErrors(GetSubscriptionName()); + _logger.EventWasNotProcessed(args.Message.MessageId); + } + } + }; + + processor.ProcessErrorAsync += ErrorHandler; + + _processors.Add(processor); + } + + private async Task EnsureSubscriptionExists(string subscriptionName) + { + if (!await _adminClient.SubscriptionExistsAsync(TOPIC_NAME, subscriptionName)) + { + _logger.LogInformation("Creating subscription on Service Bus..."); + + await _adminClient.CreateSubscriptionAsync(new CreateSubscriptionOptions(TOPIC_NAME, subscriptionName) + { + MaxDeliveryCount = MAX_DELIVERY_COUNT, + DeadLetteringOnMessageExpiration = true, + }); + + _logger.LogInformation("Successfully created subscription on Service Bus."); + } + + if (await _adminClient.RuleExistsAsync(TOPIC_NAME, subscriptionName, "$Default")) + { + await _adminClient.DeleteRuleAsync(TOPIC_NAME, subscriptionName, "$Default"); + } + } + + private async Task RegisterSubscriptionForEvent(string subscriptionName, string eventName) + { + if (!await _adminClient.RuleExistsAsync(TOPIC_NAME, subscriptionName, eventName)) + { + _logger.LogInformation("Creating rule on subscription..."); + + await _adminClient.CreateRuleAsync(TOPIC_NAME, subscriptionName, + new CreateRuleOptions + { + Filter = new CorrelationRuleFilter { Subject = eventName }, + Name = eventName + }); + + _logger.LogInformation("Successfully created rule on subscription."); + } + } + + public async Task StartConsuming(CancellationToken cancellationToken) + { + await RegisterSubscriptionClientMessageHandler(cancellationToken); + } + + private async Task RegisterSubscriptionClientMessageHandler(CancellationToken cancellationToken) + { + foreach (var processor in _processors) + { + await processor.StartProcessingAsync(cancellationToken); + } + } + + private Task ErrorHandler(ProcessErrorEventArgs args) + { + var ex = args.Exception; + var context = args.ErrorSource; + + _logger.ErrorHandlingMessage(context, ex); + + return Task.CompletedTask; + } + + private async Task ProcessEvent(string message) where TEvent : DomainEvent where THandler : IDomainEventHandler + { + var eventType = typeof(TEvent); + + var domainEvent = JsonSerializer.Deserialize(message)!; + var concreteType = typeof(IDomainEventHandler<>).MakeGenericType(eventType); + + try + { + await using var scope = _serviceProvider.CreateAsyncScope(); + + if (scope.ServiceProvider.GetService(typeof(THandler)) is not IDomainEventHandler handler) + throw new Exception($"Domain event handler '{typeof(THandler).FullName}' could not be resolved from dependency container or it does not implement {nameof(IDomainEventHandler)}."); + + var startedAt = Stopwatch.GetTimestamp(); + await (Task)concreteType.GetMethod("Handle")!.Invoke(handler, [domainEvent])!; + _metrics.TrackEventProcessingDuration(startedAt, GetSubscriptionName()); + + _metrics.IncrementNumberOfHandledEvents(GetSubscriptionName()); + } + catch (Exception ex) + { + _logger.ErrorWhileProcessingDomainEvent(domainEvent.DomainEventId, ex); + return false; + } + + return true; + } + + public async Task StopConsuming(CancellationToken cancellationToken) + { + foreach (var processor in _processors) + { + await processor.StopProcessingAsync(cancellationToken); + } + } +} diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Publisher.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Publisher.cs new file mode 100644 index 0000000000..7ae0330314 --- /dev/null +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.Publisher.cs @@ -0,0 +1,47 @@ +using System.Diagnostics; +using System.Text; +using System.Text.Json; +using Azure.Messaging.ServiceBus; +using Backbone.BuildingBlocks.Domain.Events; +using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; +using Microsoft.Extensions.Logging; + +namespace Backbone.BuildingBlocks.Infrastructure.EventBus.AzureServiceBus; + +public partial class EventBusAzureServiceBus +{ + public async Task Publish(DomainEvent @event) + { + var eventName = @event.GetEventName(); + var jsonMessage = JsonSerializer.Serialize(@event, @event.GetType()); + var body = Encoding.UTF8.GetBytes(jsonMessage); + + _metrics.TrackHandledMessageSize(body.Length); + + var message = new ServiceBusMessage + { + MessageId = @event.DomainEventId, + Body = new BinaryData(body), + Subject = eventName, + CorrelationId = CustomLogContext.GetCorrelationId() + }; + + _logger.SendingDomainEvent(message.MessageId); + + try + { + var startedAt = Stopwatch.GetTimestamp(); + await _sender.SendMessageAsync(message); + _metrics.TrackEventPublishingDuration(startedAt); + + _metrics.IncrementNumberOfPublishedEvents(eventName); + + _logger.LogDebug("Successfully sent domain event with id '{MessageId}'.", message.MessageId); + } + catch (Exception) + { + _metrics.IncrementNumberOfPublishingErrors(eventName); + throw; + } + } +} diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.cs index 8e6a090226..17806d957a 100644 --- a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.cs +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/AzureServiceBus/EventBusAzureServiceBus.cs @@ -1,18 +1,13 @@ -using System.Diagnostics; -using System.Text; -using System.Text.Json; using Azure.Messaging.ServiceBus; using Azure.Messaging.ServiceBus.Administration; using Backbone.BuildingBlocks.Application.Abstractions.Infrastructure.EventBus; using Backbone.BuildingBlocks.Domain.Events; -using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; using Backbone.Tooling.Extensions; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; namespace Backbone.BuildingBlocks.Infrastructure.EventBus.AzureServiceBus; -public class EventBusAzureServiceBus : IEventBus, IDisposable, IAsyncDisposable +public partial class EventBusAzureServiceBus : IEventBus, IDisposable, IAsyncDisposable { private const string TOPIC_NAME = "default"; private const int MAX_DELIVERY_COUNT = 5; @@ -57,182 +52,8 @@ public async ValueTask DisposeAsync() } } - public async Task Publish(DomainEvent @event) - { - var eventName = @event.GetEventName(); - var jsonMessage = JsonSerializer.Serialize(@event, @event.GetType()); - var body = Encoding.UTF8.GetBytes(jsonMessage); - - _metrics.TrackHandledMessageSize(body.Length); - - var message = new ServiceBusMessage - { - MessageId = @event.DomainEventId, - Body = new BinaryData(body), - Subject = eventName, - CorrelationId = CustomLogContext.GetCorrelationId() - }; - - _logger.SendingDomainEvent(message.MessageId); - - try - { - var startedAt = Stopwatch.GetTimestamp(); - await _sender.SendMessageAsync(message); - _metrics.TrackEventPublishingDuration(startedAt); - - _metrics.IncrementNumberOfPublishedEvents(eventName); - - _logger.LogDebug("Successfully sent domain event with id '{MessageId}'.", message.MessageId); - } - catch (Exception) - { - _metrics.IncrementNumberOfPublishingErrors(eventName); - throw; - } - } - - public async Task Subscribe() - where T : DomainEvent - where TH : IDomainEventHandler - { - var eventName = typeof(T).GetEventName(); - var subscriptionName = GetSubscriptionName(); - - await EnsureSubscriptionExists(subscriptionName); - - await RegisterSubscriptionForEvent(subscriptionName, eventName); - - var processor = _client.CreateProcessor(TOPIC_NAME, subscriptionName, _options); - - processor.ProcessMessageAsync += - async args => - { - var messageData = args.Message.Body.ToString(); - var correlationId = args.Message.CorrelationId; - - correlationId = correlationId.IsNullOrEmpty() ? CustomLogContext.GenerateCorrelationId() : correlationId; - - using (CustomLogContext.SetCorrelationId(correlationId)) - { - var processedSuccessfully = await ProcessEvent(messageData); - - if (processedSuccessfully) - await args.CompleteMessageAsync(args.Message); - else - { - await args.AbandonMessageAsync(args.Message); - _metrics.IncrementNumberOfProcessingErrors(GetSubscriptionName()); - _logger.EventWasNotProcessed(args.Message.MessageId); - } - } - }; - - processor.ProcessErrorAsync += ErrorHandler; - - _processors.Add(processor); - } - public bool IsConnected => !_client.IsClosed; - private async Task EnsureSubscriptionExists(string subscriptionName) - { - if (!await _adminClient.SubscriptionExistsAsync(TOPIC_NAME, subscriptionName)) - { - _logger.LogInformation("Creating subscription on Service Bus..."); - - await _adminClient.CreateSubscriptionAsync(new CreateSubscriptionOptions(TOPIC_NAME, subscriptionName) - { - MaxDeliveryCount = MAX_DELIVERY_COUNT, - DeadLetteringOnMessageExpiration = true, - }); - - _logger.LogInformation("Successfully created subscription on Service Bus."); - } - - if (await _adminClient.RuleExistsAsync(TOPIC_NAME, subscriptionName, "$Default")) - { - await _adminClient.DeleteRuleAsync(TOPIC_NAME, subscriptionName, "$Default"); - } - } - - private async Task RegisterSubscriptionForEvent(string subscriptionName, string eventName) - { - if (!await _adminClient.RuleExistsAsync(TOPIC_NAME, subscriptionName, eventName)) - { - _logger.LogInformation("Creating rule on subscription..."); - - await _adminClient.CreateRuleAsync(TOPIC_NAME, subscriptionName, - new CreateRuleOptions - { - Filter = new CorrelationRuleFilter { Subject = eventName }, - Name = eventName - }); - - _logger.LogInformation("Successfully created rule on subscription."); - } - } - - public async Task StartConsuming(CancellationToken cancellationToken) - { - await RegisterSubscriptionClientMessageHandler(cancellationToken); - } - - private async Task RegisterSubscriptionClientMessageHandler(CancellationToken cancellationToken) - { - foreach (var processor in _processors) - { - await processor.StartProcessingAsync(cancellationToken); - } - } - - private Task ErrorHandler(ProcessErrorEventArgs args) - { - var ex = args.Exception; - var context = args.ErrorSource; - - _logger.ErrorHandlingMessage(context, ex); - - return Task.CompletedTask; - } - - private async Task ProcessEvent(string message) where TEvent : DomainEvent where THandler : IDomainEventHandler - { - var eventType = typeof(TEvent); - - var domainEvent = JsonSerializer.Deserialize(message)!; - var concreteType = typeof(IDomainEventHandler<>).MakeGenericType(eventType); - - try - { - await using var scope = _serviceProvider.CreateAsyncScope(); - - if (scope.ServiceProvider.GetService(typeof(THandler)) is not IDomainEventHandler handler) - throw new Exception($"Domain event handler '{typeof(THandler).FullName}' could not be resolved from dependency container or it does not implement {nameof(IDomainEventHandler)}."); - - var startedAt = Stopwatch.GetTimestamp(); - await (Task)concreteType.GetMethod("Handle")!.Invoke(handler, [domainEvent])!; - _metrics.TrackEventProcessingDuration(startedAt, GetSubscriptionName()); - - _metrics.IncrementNumberOfHandledEvents(GetSubscriptionName()); - } - catch (Exception ex) - { - _logger.ErrorWhileProcessingDomainEvent(domainEvent.DomainEventId, ex); - return false; - } - - return true; - } - - public async Task StopConsuming(CancellationToken cancellationToken) - { - foreach (var processor in _processors) - { - await processor.StopProcessingAsync(cancellationToken); - } - } - public static string GetSubscriptionName() where TEvent : DomainEvent where THandler : IDomainEventHandler { var eventHandlerFullName = typeof(THandler).FullName!; From bed92b8c169a1eaaa2eb9e33dd4f78a5dc3d6c84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Timo=20H=C3=BCgel?= Date: Wed, 26 Aug 2026 09:49:06 +0200 Subject: [PATCH 3/3] feat: introduce partial classes for GoogleCloudPubSub --- .../EventBusGoogleCloudPubSub.Consumer.cs | 130 +++++++++++++++ .../EventBusGoogleCloudPubSub.Publisher.cs | 45 +++++ .../EventBusGoogleCloudPubSub.cs | 155 +----------------- 3 files changed, 176 insertions(+), 154 deletions(-) create mode 100644 BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Consumer.cs create mode 100644 BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Publisher.cs diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Consumer.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Consumer.cs new file mode 100644 index 0000000000..078136fdfb --- /dev/null +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Consumer.cs @@ -0,0 +1,130 @@ +using System.Diagnostics; +using System.Text.Json; +using Backbone.BuildingBlocks.Application.Abstractions.Infrastructure.EventBus; +using Backbone.BuildingBlocks.Domain.Events; +using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; +using Backbone.Tooling.Extensions; +using Google.Api.Gax; +using Google.Cloud.PubSub.V1; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Type = System.Type; + +namespace Backbone.BuildingBlocks.Infrastructure.EventBus.GoogleCloudPubSub; + +public partial class EventBusGoogleCloudPubSub +{ + public async Task Subscribe() + where T : DomainEvent + where TH : IDomainEventHandler + { + var eventName = typeof(T).GetEventName(); + var subscriptionName = GetSubscriptionName(_projectId); + + await EnsureSubscriptionExists(subscriptionName, eventName); + + var subscriberClient = await new SubscriberClientBuilder + { + SubscriptionName = subscriptionName, + GoogleCredential = _gcpCredentials, + EmulatorDetection = EmulatorDetection.EmulatorOrProduction, + }.BuildAsync(); + + _subscriptions.Add(new Subscription(subscriberClient, typeof(T), typeof(TH))); + } + + private async Task EnsureSubscriptionExists(SubscriptionName subscriptionName, string eventName) + { + try + { + var subscriptionRequest = new Google.Cloud.PubSub.V1.Subscription + { + SubscriptionName = subscriptionName, + TopicAsTopicName = _topicName, + Filter = $"attributes.{PubSubMessageAttributes.EVENT_NAME} = \"{eventName}\"", + AckDeadlineSeconds = (int)MESSAGE_ACK_DEADLINE.TotalSeconds, + RetryPolicy = new RetryPolicy + { + MinimumBackoff = Duration.FromTimeSpan(SUBSCRIPTION_MINIMUM_BACKOFF.Seconds()), + MaximumBackoff = Duration.FromTimeSpan(SUBSCRIPTION_MAXIMUM_BACKOFF.Seconds()) + } + }; + + _logger.LogInformation("Creating subscription '{SubscriptionName}' for event '{EventName}'...", subscriptionName, eventName); + + await _subscriberService.CreateSubscriptionAsync(subscriptionRequest); + + _logger.LogInformation("Successfully created subscription '{SubscriptionName}' for event '{EventName}'.", subscriptionName, eventName); + } + catch (RpcException ex) + { + if (ex.StatusCode == StatusCode.AlreadyExists) + { + _logger.LogInformation("Subscription '{SubscriptionName}' for event '{EventName}' already exists.", subscriptionName, eventName); + return; + } + + throw; + } + } + + public async Task StartConsuming(CancellationToken cancellationToken) + { + var consumptionTasks = _subscriptions.Select(s => s.SubscriberClient.StartAsync((e, _) => OnIncomingEvent(e, s.EventType, s.HandlerType))); + + await Task.WhenAll(consumptionTasks); + } + + private async Task OnIncomingEvent(PubsubMessage @event, Type eventType, Type handlerType) + { + var eventData = @event.Data.ToStringUtf8(); + + try + { + @event.Attributes.TryGetValue(PubSubMessageAttributes.CORRELATION_ID, out var correlationId); + + correlationId = correlationId.IsNullOrEmpty() ? CustomLogContext.GenerateCorrelationId() : correlationId; + + using (CustomLogContext.SetCorrelationId(correlationId)) + { + await ProcessEvent(eventData, eventType, handlerType); + } + } + catch (Exception ex) + { + _metrics.IncrementNumberOfProcessingErrors(GetSubscriptionName(_projectId, handlerType, eventType).SubscriptionId); + _logger.ErrorHandlingMessage(ex.StackTrace!, ex); + return SubscriberClient.Reply.Nack; + } + + return SubscriberClient.Reply.Ack; + } + + private async Task ProcessEvent(string message, Type eventType, Type handlerType) + { + var subscriptionName = GetSubscriptionName(_projectId, handlerType, eventType).SubscriptionId; + var domainEvent = JsonSerializer.Deserialize(message, eventType)!; + + await using var scope = _serviceProvider.CreateAsyncScope(); + + if (scope.ServiceProvider.GetService(handlerType) is not IDomainEventHandler handler) + throw new Exception("Domain event handler could not be resolved from dependency container or it does not implement IDomainEventHandler."); + + var handleMethod = handler.GetType().GetMethod("Handle"); + + var startedAt = Stopwatch.GetTimestamp(); + await (Task)handleMethod!.Invoke(handler, [domainEvent])!; + _metrics.TrackEventProcessingDuration(startedAt, subscriptionName); + + _metrics.IncrementNumberOfHandledEvents(subscriptionName); + } + + public async Task StopConsuming(CancellationToken cancellationToken) + { + var stopTasks = _subscriptions.Select(subscription => subscription.SubscriberClient.StopAsync(new SubscriberClient.ShutdownOptions(), CancellationToken.None)); + + await Task.WhenAll(stopTasks); + } +} diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Publisher.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Publisher.cs new file mode 100644 index 0000000000..647efdccb5 --- /dev/null +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.Publisher.cs @@ -0,0 +1,45 @@ +using System.Diagnostics; +using System.Text.Json; +using Backbone.BuildingBlocks.Domain.Events; +using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; +using Google.Cloud.PubSub.V1; +using Google.Protobuf; + +namespace Backbone.BuildingBlocks.Infrastructure.EventBus.GoogleCloudPubSub; + +public partial class EventBusGoogleCloudPubSub +{ + public async Task Publish(DomainEvent @event) + { + var eventName = @event.GetEventName(); + var jsonMessage = JsonSerializer.Serialize(@event, @event.GetType()); + var messageBytes = ByteString.CopyFromUtf8(jsonMessage); + + _metrics.TrackHandledMessageSize(messageBytes.Length); + + var message = new PubsubMessage + { + Data = messageBytes, + Attributes = + { + { PubSubMessageAttributes.EVENT_NAME, eventName }, + { PubSubMessageAttributes.CORRELATION_ID, CustomLogContext.GetCorrelationId() } + } + }; + + try + { + var startedAt = Stopwatch.GetTimestamp(); + var messageId = await _publisherClient.PublishAsync(message); + _logger.SuccessfullySentDomainEvent(messageId); + + _metrics.TrackEventPublishingDuration(startedAt); + _metrics.IncrementNumberOfPublishedEvents(eventName); + } + catch (Exception) + { + _metrics.IncrementNumberOfPublishingErrors(eventName); + throw; + } + } +} diff --git a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.cs b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.cs index bc2785fb46..79e657b108 100644 --- a/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.cs +++ b/BuildingBlocks/src/BuildingBlocks.Infrastructure/EventBus/GoogleCloudPubSub/EventBusGoogleCloudPubSub.cs @@ -1,22 +1,15 @@ -using System.Diagnostics; -using System.Text.Json; using Backbone.BuildingBlocks.Application.Abstractions.Infrastructure.EventBus; using Backbone.BuildingBlocks.Domain.Events; -using Backbone.BuildingBlocks.Infrastructure.CorrelationIds; using Backbone.Tooling.Extensions; using Google.Api.Gax; using Google.Apis.Auth.OAuth2; using Google.Cloud.PubSub.V1; -using Google.Protobuf; -using Google.Protobuf.WellKnownTypes; -using Grpc.Core; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Type = System.Type; namespace Backbone.BuildingBlocks.Infrastructure.EventBus.GoogleCloudPubSub; -public class EventBusGoogleCloudPubSub : IEventBus, IDisposable, IAsyncDisposable +public partial class EventBusGoogleCloudPubSub : IEventBus, IDisposable, IAsyncDisposable { private const int SUBSCRIPTION_MINIMUM_BACKOFF = 2; private const int SUBSCRIPTION_MAXIMUM_BACKOFF = 120; @@ -64,155 +57,9 @@ public EventBusGoogleCloudPubSub(ILogger logger, ISer }.Build(); } - public async Task Publish(DomainEvent @event) - { - var eventName = @event.GetEventName(); - var jsonMessage = JsonSerializer.Serialize(@event, @event.GetType()); - var messageBytes = ByteString.CopyFromUtf8(jsonMessage); - - _metrics.TrackHandledMessageSize(messageBytes.Length); - - var message = new PubsubMessage - { - Data = messageBytes, - Attributes = - { - { PubSubMessageAttributes.EVENT_NAME, eventName }, - { PubSubMessageAttributes.CORRELATION_ID, CustomLogContext.GetCorrelationId() } - } - }; - - try - { - var startedAt = Stopwatch.GetTimestamp(); - var messageId = await _publisherClient.PublishAsync(message); - _logger.SuccessfullySentDomainEvent(messageId); - - _metrics.TrackEventPublishingDuration(startedAt); - _metrics.IncrementNumberOfPublishedEvents(eventName); - } - catch (Exception) - { - _metrics.IncrementNumberOfPublishingErrors(eventName); - throw; - } - } - - public async Task Subscribe() - where T : DomainEvent - where TH : IDomainEventHandler - { - var eventName = typeof(T).GetEventName(); - var subscriptionName = GetSubscriptionName(_projectId); - - await EnsureSubscriptionExists(subscriptionName, eventName); - - var subscriberClient = await new SubscriberClientBuilder - { - SubscriptionName = subscriptionName, - GoogleCredential = _gcpCredentials, - EmulatorDetection = EmulatorDetection.EmulatorOrProduction, - }.BuildAsync(); - - _subscriptions.Add(new Subscription(subscriberClient, typeof(T), typeof(TH))); - } - // We currently don't know how to properly implement this for Pub/Sub. So for now, we just return true. public bool IsConnected => true; - private async Task EnsureSubscriptionExists(SubscriptionName subscriptionName, string eventName) - { - try - { - var subscriptionRequest = new Google.Cloud.PubSub.V1.Subscription - { - SubscriptionName = subscriptionName, - TopicAsTopicName = _topicName, - Filter = $"attributes.{PubSubMessageAttributes.EVENT_NAME} = \"{eventName}\"", - AckDeadlineSeconds = (int)MESSAGE_ACK_DEADLINE.TotalSeconds, - RetryPolicy = new RetryPolicy - { - MinimumBackoff = Duration.FromTimeSpan(SUBSCRIPTION_MINIMUM_BACKOFF.Seconds()), - MaximumBackoff = Duration.FromTimeSpan(SUBSCRIPTION_MAXIMUM_BACKOFF.Seconds()) - } - }; - - _logger.LogInformation("Creating subscription '{SubscriptionName}' for event '{EventName}'...", subscriptionName, eventName); - - await _subscriberService.CreateSubscriptionAsync(subscriptionRequest); - - _logger.LogInformation("Successfully created subscription '{SubscriptionName}' for event '{EventName}'.", subscriptionName, eventName); - } - catch (RpcException ex) - { - if (ex.StatusCode == StatusCode.AlreadyExists) - { - _logger.LogInformation("Subscription '{SubscriptionName}' for event '{EventName}' already exists.", subscriptionName, eventName); - return; - } - - throw; - } - } - - public async Task StartConsuming(CancellationToken cancellationToken) - { - var consumptionTasks = _subscriptions.Select(s => s.SubscriberClient.StartAsync((e, _) => OnIncomingEvent(e, s.EventType, s.HandlerType))); - - await Task.WhenAll(consumptionTasks); - } - - private async Task OnIncomingEvent(PubsubMessage @event, Type eventType, Type handlerType) - { - var eventData = @event.Data.ToStringUtf8(); - - try - { - @event.Attributes.TryGetValue(PubSubMessageAttributes.CORRELATION_ID, out var correlationId); - - correlationId = correlationId.IsNullOrEmpty() ? CustomLogContext.GenerateCorrelationId() : correlationId; - - using (CustomLogContext.SetCorrelationId(correlationId)) - { - await ProcessEvent(eventData, eventType, handlerType); - } - } - catch (Exception ex) - { - _metrics.IncrementNumberOfProcessingErrors(GetSubscriptionName(_projectId, handlerType, eventType).SubscriptionId); - _logger.ErrorHandlingMessage(ex.StackTrace!, ex); - return SubscriberClient.Reply.Nack; - } - - return SubscriberClient.Reply.Ack; - } - - private async Task ProcessEvent(string message, Type eventType, Type handlerType) - { - var subscriptionName = GetSubscriptionName(_projectId, handlerType, eventType).SubscriptionId; - var domainEvent = JsonSerializer.Deserialize(message, eventType)!; - - await using var scope = _serviceProvider.CreateAsyncScope(); - - if (scope.ServiceProvider.GetService(handlerType) is not IDomainEventHandler handler) - throw new Exception("Domain event handler could not be resolved from dependency container or it does not implement IDomainEventHandler."); - - var handleMethod = handler.GetType().GetMethod("Handle"); - - var startedAt = Stopwatch.GetTimestamp(); - await (Task)handleMethod!.Invoke(handler, [domainEvent])!; - _metrics.TrackEventProcessingDuration(startedAt, subscriptionName); - - _metrics.IncrementNumberOfHandledEvents(subscriptionName); - } - - public async Task StopConsuming(CancellationToken cancellationToken) - { - var stopTasks = _subscriptions.Select(subscription => subscription.SubscriberClient.StopAsync(new SubscriberClient.ShutdownOptions(), CancellationToken.None)); - - await Task.WhenAll(stopTasks); - } - public void Dispose() { Task.Run(async () => await DisposeAsync()).GetAwaiter().GetResult();