From d31e2ba3d731a52ad9d9c8ad81c6316967d263ff Mon Sep 17 00:00:00 2001 From: MilePrivate Date: Wed, 5 Aug 2026 19:12:18 +0200 Subject: [PATCH] Publish outbox events to RabbitMQ --- .../Unit/RabbitMqTopologyTests.cs | 56 +++++ .../OrderProcessing.Api.csproj | 1 + OrderProcessing.Api/Program.cs | 60 ++++- .../RabbitMqIntegrationEventPublisher.cs | 231 ++++++++++++++++++ .../Services/Messaging/RabbitMqOptions.cs | 26 ++ .../Services/Messaging/RabbitMqTopology.cs | 34 +++ .../appsettings.Development.json | 3 + OrderProcessing.Api/appsettings.json | 12 + 8 files changed, 422 insertions(+), 1 deletion(-) create mode 100644 OrderProcessing.Api.Tests/Unit/RabbitMqTopologyTests.cs create mode 100644 OrderProcessing.Api/Services/Messaging/RabbitMqIntegrationEventPublisher.cs create mode 100644 OrderProcessing.Api/Services/Messaging/RabbitMqOptions.cs create mode 100644 OrderProcessing.Api/Services/Messaging/RabbitMqTopology.cs diff --git a/OrderProcessing.Api.Tests/Unit/RabbitMqTopologyTests.cs b/OrderProcessing.Api.Tests/Unit/RabbitMqTopologyTests.cs new file mode 100644 index 0000000..c23beb2 --- /dev/null +++ b/OrderProcessing.Api.Tests/Unit/RabbitMqTopologyTests.cs @@ -0,0 +1,56 @@ +using OrderProcessing.Api.Services.Messaging; +using OrderProcessing.Contracts.Orders; + +namespace OrderProcessing.Api.Tests.Unit; + +public sealed class RabbitMqTopologyTests +{ + [Fact] + public void GetRoutingKey_ForOrderCreatedEvent_ReturnsCreatedKey() + { + var routingKey = RabbitMqTopology.GetRoutingKey( + typeof(OrderCreatedIntegrationEvent).FullName!); + + Assert.Equal( + RabbitMqTopology.OrderCreatedRoutingKey, + routingKey); + } + + [Fact] + public void GetRoutingKey_ForOrderCompletedEvent_ReturnsCompletedKey() + { + var routingKey = RabbitMqTopology.GetRoutingKey( + typeof(OrderCompletedIntegrationEvent).FullName!); + + Assert.Equal( + RabbitMqTopology.OrderCompletedRoutingKey, + routingKey); + } + + [Fact] + public void GetRoutingKey_ForOrderCancelledEvent_ReturnsCancelledKey() + { + var routingKey = RabbitMqTopology.GetRoutingKey( + typeof(OrderCancelledIntegrationEvent).FullName!); + + Assert.Equal( + RabbitMqTopology.OrderCancelledRoutingKey, + routingKey); + } + + [Fact] + public void GetRoutingKey_ForUnknownEvent_ThrowsException() + { + var action = () => + RabbitMqTopology.GetRoutingKey( + "UnknownIntegrationEvent"); + + var exception = + Assert.Throws( + action); + + Assert.Contains( + "UnknownIntegrationEvent", + exception.Message); + } +} \ No newline at end of file diff --git a/OrderProcessing.Api/OrderProcessing.Api.csproj b/OrderProcessing.Api/OrderProcessing.Api.csproj index 19c71c5..b6a871d 100644 --- a/OrderProcessing.Api/OrderProcessing.Api.csproj +++ b/OrderProcessing.Api/OrderProcessing.Api.csproj @@ -20,6 +20,7 @@ runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/OrderProcessing.Api/Program.cs b/OrderProcessing.Api/Program.cs index f25fb7e..01355a8 100644 --- a/OrderProcessing.Api/Program.cs +++ b/OrderProcessing.Api/Program.cs @@ -77,10 +77,68 @@ public static async Task Main(string[] args) "Outbox polling interval must be greater than zero.") .ValidateOnStart(); - builder.Services.AddSingleton(); + var rabbitMqEnabled = builder.Configuration + .GetSection(RabbitMqOptions.SectionName) + .GetValue( + nameof(RabbitMqOptions.Enabled)); + + if (rabbitMqEnabled && !builder.Environment.IsEnvironment("Testing")) + { + builder.Services.AddSingleton< + IIntegrationEventPublisher, + RabbitMqIntegrationEventPublisher>(); + } + else + { + builder.Services.AddSingleton< + IIntegrationEventPublisher, + LoggingIntegrationEventPublisher>(); + } builder.Services.AddScoped(); + builder.Services + .AddOptions() + .Bind( + builder.Configuration.GetSection( + RabbitMqOptions.SectionName)) + .Validate( + options => + !options.Enabled || + !string.IsNullOrWhiteSpace( + options.HostName), + "RabbitMQ host name is required when RabbitMQ is enabled.") + .Validate( + options => + !options.Enabled || + options.Port > 0, + "RabbitMQ port must be greater than zero.") + .Validate( + options => + !options.Enabled || + !string.IsNullOrWhiteSpace( + options.UserName), + "RabbitMQ user name is required when RabbitMQ is enabled.") + .Validate( + options => + !options.Enabled || + !string.IsNullOrWhiteSpace( + options.Password), + "RabbitMQ password is required when RabbitMQ is enabled.") + .Validate( + options => + !options.Enabled || + !string.IsNullOrWhiteSpace( + options.ExchangeName), + "RabbitMQ exchange name is required when RabbitMQ is enabled.") + .Validate( + options => + !options.Enabled || + !string.IsNullOrWhiteSpace( + options.EmailQueueName), + "RabbitMQ email queue name is required when RabbitMQ is enabled.") + .ValidateOnStart(); + var app = builder.Build(); if (app.Environment.IsDevelopment()) diff --git a/OrderProcessing.Api/Services/Messaging/RabbitMqIntegrationEventPublisher.cs b/OrderProcessing.Api/Services/Messaging/RabbitMqIntegrationEventPublisher.cs new file mode 100644 index 0000000..b716fdf --- /dev/null +++ b/OrderProcessing.Api/Services/Messaging/RabbitMqIntegrationEventPublisher.cs @@ -0,0 +1,231 @@ +using System.Text; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; + +namespace OrderProcessing.Api.Services.Messaging; + +public sealed class RabbitMqIntegrationEventPublisher : IIntegrationEventPublisher, IAsyncDisposable +{ + private readonly RabbitMqOptions _options; + private readonly ConnectionFactory _connectionFactory; + private readonly ILogger _logger; + + private readonly SemaphoreSlim _publishLock = new(initialCount: 1, maxCount: 1); + + private IConnection? _connection; + private IChannel? _channel; + + public RabbitMqIntegrationEventPublisher(IOptions options, ILogger logger) + { + _options = options.Value; + _logger = logger; + _connectionFactory = new ConnectionFactory + { + HostName = _options.HostName, + Port = _options.Port, + UserName = _options.UserName, + Password = _options.Password, + VirtualHost = _options.VirtualHost, + + AutomaticRecoveryEnabled = true, + TopologyRecoveryEnabled = true, + + NetworkRecoveryInterval = + TimeSpan.FromSeconds(_options.NetworkRecoveryIntervalSeconds) + }; + } + + public async Task PublishAsync(IntegrationEventEnvelope message, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(message); + + await _publishLock.WaitAsync(cancellationToken); + + try + { + var channel = await GetOrCreateChannelAsync(cancellationToken); + + var routingKey =RabbitMqTopology.GetRoutingKey(message.Type); + + var properties = new BasicProperties + { + ContentType = "application/json", + ContentEncoding = "utf-8", + MessageId = message.MessageId.ToString(), + Type = message.Type, + AppId = "OrderProcessing.Api", + Persistent = true, + Timestamp = new AmqpTimestamp( + new DateTimeOffset( + message.OccurredAtUtc) + .ToUnixTimeSeconds()) + }; + + var body = Encoding.UTF8.GetBytes( + message.Payload); + + await channel.BasicPublishAsync( + exchange: _options.ExchangeName, + routingKey: routingKey, + mandatory: true, + basicProperties: properties, + body: body, + cancellationToken: cancellationToken); + + _logger.LogInformation( + "Published integration event {MessageId} " + + "with type {EventType} and routing key {RoutingKey}", + message.MessageId, + message.Type, + routingKey); + } + catch + { + await ResetConnectionAsync(); + throw; + } + finally + { + _publishLock.Release(); + } + } + + private async Task GetOrCreateChannelAsync(CancellationToken cancellationToken) + { + if (_channel is { IsOpen: true }) + { + return _channel; + } + + await ResetConnectionAsync(); + + _connection = + await _connectionFactory.CreateConnectionAsync( + _options.ClientProvidedName, + cancellationToken); + + var channelOptions = new CreateChannelOptions( + publisherConfirmationsEnabled: true, + publisherConfirmationTrackingEnabled: true); + + _channel = + await _connection.CreateChannelAsync( + channelOptions, + cancellationToken); + + await DeclareTopologyAsync( + _channel, + cancellationToken); + + _logger.LogInformation( + "RabbitMQ publisher connected to {HostName}:{Port} " + + "using exchange {ExchangeName}", + _options.HostName, + _options.Port, + _options.ExchangeName); + + return _channel; + } + + private async Task DeclareTopologyAsync(IChannel channel, CancellationToken cancellationToken) + { + await channel.ExchangeDeclareAsync( + exchange: _options.ExchangeName, + type: ExchangeType.Topic, + durable: true, + autoDelete: false, + cancellationToken: cancellationToken); + + await channel.QueueDeclareAsync( + queue: _options.EmailQueueName, + durable: true, + exclusive: false, + autoDelete: false, + cancellationToken: cancellationToken); + + await channel.QueueBindAsync( + queue: _options.EmailQueueName, + exchange: _options.ExchangeName, + routingKey: + RabbitMqTopology.OrderCreatedRoutingKey, + cancellationToken: cancellationToken); + + await channel.QueueBindAsync( + queue: _options.EmailQueueName, + exchange: _options.ExchangeName, + routingKey: + RabbitMqTopology.OrderCompletedRoutingKey, + cancellationToken: cancellationToken); + + await channel.QueueBindAsync( + queue: _options.EmailQueueName, + exchange: _options.ExchangeName, + routingKey: + RabbitMqTopology.OrderCancelledRoutingKey, + cancellationToken: cancellationToken); + } + + private async ValueTask ResetConnectionAsync() + { + var channel = _channel; + _channel = null; + + if (channel is not null) + { + try + { + if (channel.IsOpen) + { + await channel.CloseAsync(CancellationToken.None); + } + } + catch (Exception exception) + { + _logger.LogDebug( + exception, + "An error occurred while closing " + + "the RabbitMQ channel"); + } + + await channel.DisposeAsync(); + } + + var connection = _connection; + _connection = null; + + if (connection is not null) + { + try + { + if (connection.IsOpen) + { + await connection.CloseAsync(CancellationToken.None); + } + } + catch (Exception exception) + { + _logger.LogDebug( + exception, + "An error occurred while closing " + + "the RabbitMQ connection"); + } + + await connection.DisposeAsync(); + } + } + + public async ValueTask DisposeAsync() + { + await _publishLock.WaitAsync(); + + try + { + await ResetConnectionAsync(); + } + finally + { + _publishLock.Release(); + _publishLock.Dispose(); + } + } +} \ No newline at end of file diff --git a/OrderProcessing.Api/Services/Messaging/RabbitMqOptions.cs b/OrderProcessing.Api/Services/Messaging/RabbitMqOptions.cs new file mode 100644 index 0000000..ccb95c6 --- /dev/null +++ b/OrderProcessing.Api/Services/Messaging/RabbitMqOptions.cs @@ -0,0 +1,26 @@ +namespace OrderProcessing.Api.Services.Messaging; + +public sealed class RabbitMqOptions +{ + public const string SectionName = "RabbitMq"; + + public bool Enabled { get; set; } + + public string HostName { get; set; } = "localhost"; + + public int Port { get; set; } = 5672; + + public string UserName { get; set; } = "guest"; + + public string Password { get; set; } = "guest"; + + public string VirtualHost { get; set; } = "/"; + + public string ExchangeName { get; set; } = "order-processing.events"; + + public string EmailQueueName { get; set; } = "order-processing.email"; + + public string ClientProvidedName { get; set; } = "order-processing-api-publisher"; + + public int NetworkRecoveryIntervalSeconds { get; set; } = 5; +} \ No newline at end of file diff --git a/OrderProcessing.Api/Services/Messaging/RabbitMqTopology.cs b/OrderProcessing.Api/Services/Messaging/RabbitMqTopology.cs new file mode 100644 index 0000000..c282059 --- /dev/null +++ b/OrderProcessing.Api/Services/Messaging/RabbitMqTopology.cs @@ -0,0 +1,34 @@ +using OrderProcessing.Contracts.Orders; + +namespace OrderProcessing.Api.Services.Messaging; + +public static class RabbitMqTopology +{ + public const string OrderCreatedRoutingKey = "order.created"; + + public const string OrderCompletedRoutingKey = "order.completed"; + + public const string OrderCancelledRoutingKey = "order.cancelled"; + + public static string GetRoutingKey(string eventType) + { + return eventType switch + { + var type when type == + typeof(OrderCreatedIntegrationEvent).FullName => + OrderCreatedRoutingKey, + + var type when type == + typeof(OrderCompletedIntegrationEvent).FullName => + OrderCompletedRoutingKey, + + var type when type == + typeof(OrderCancelledIntegrationEvent).FullName => + OrderCancelledRoutingKey, + + _ => throw new InvalidOperationException( + $"No RabbitMQ routing key is configured " + + $"for integration event type '{eventType}'.") + }; + } +} \ No newline at end of file diff --git a/OrderProcessing.Api/appsettings.Development.json b/OrderProcessing.Api/appsettings.Development.json index 0c208ae..b00282e 100644 --- a/OrderProcessing.Api/appsettings.Development.json +++ b/OrderProcessing.Api/appsettings.Development.json @@ -3,6 +3,9 @@ "LogLevel": { "Default": "Information", "Microsoft.AspNetCore": "Warning" + }, + "RabbitMq": { + "Enabled": true } } } diff --git a/OrderProcessing.Api/appsettings.json b/OrderProcessing.Api/appsettings.json index a17eb64..4ebe231 100644 --- a/OrderProcessing.Api/appsettings.json +++ b/OrderProcessing.Api/appsettings.json @@ -42,5 +42,17 @@ "BatchSize": 20, "MaxRetryCount": 5, "PollingIntervalSeconds": 5 + }, + "RabbitMq": { + "Enabled": false, + "HostName": "localhost", + "Port": 5672, + "UserName": "guest", + "Password": "guest", + "VirtualHost": "/", + "ExchangeName": "order-processing.events", + "EmailQueueName": "order-processing.email", + "ClientProvidedName": "order-processing-api-publisher", + "NetworkRecoveryIntervalSeconds": 5 } }