Publish outbox events to RabbitMQ

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<InvalidOperationException>(
+                action);
+
+        Assert.Contains(
+            "UnknownIntegrationEvent",
+            exception.Message);
+    }
+}

OrderProcessing.Api/OrderProcessing.Api.csproj

@@ -20,6 +20,7 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
     <PackageReference Include="Microsoft.OpenApi" Version="2.7.5" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
     <PackageReference Include="Serilog.AspNetCore" Version="10.0.0" />
     <PackageReference Include="Serilog.Settings.Configuration" Version="10.0.1" />
     <PackageReference Include="Serilog.Sinks.Console" Version="6.1.1" />

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<IIntegrationEventPublisher, LoggingIntegrationEventPublisher>();
+            var rabbitMqEnabled = builder.Configuration
+            .GetSection(RabbitMqOptions.SectionName)
+            .GetValue<bool>(
+                nameof(RabbitMqOptions.Enabled));
+
+                    if (rabbitMqEnabled && !builder.Environment.IsEnvironment("Testing"))
+                    {
+                        builder.Services.AddSingleton<
+                            IIntegrationEventPublisher,
+                            RabbitMqIntegrationEventPublisher>();
+                    }
+                    else
+                    {
+                        builder.Services.AddSingleton<
+                            IIntegrationEventPublisher,
+                            LoggingIntegrationEventPublisher>();
+                    }
 
             builder.Services.AddScoped<OutboxProcessor>();
 
+            builder.Services
+           .AddOptions<RabbitMqOptions>()
+           .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())

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<RabbitMqIntegrationEventPublisher> _logger;
+
+    private readonly SemaphoreSlim _publishLock = new(initialCount: 1, maxCount: 1);
+
+    private IConnection? _connection;
+    private IChannel? _channel;
+
+    public RabbitMqIntegrationEventPublisher(IOptions<RabbitMqOptions> options, ILogger<RabbitMqIntegrationEventPublisher> 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<IChannel> 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();
+        }
+    }
+}

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;
+}