diff --git a/src/infrastructure/DigitalData.MessagingService.Publisher/DependencyInjection.cs b/src/infrastructure/DigitalData.MessagingService.Publisher/DependencyInjection.cs new file mode 100644 index 0000000..e93a681 --- /dev/null +++ b/src/infrastructure/DigitalData.MessagingService.Publisher/DependencyInjection.cs @@ -0,0 +1,38 @@ +using DigitalData.MessagingService.Publisher.Abstraction; +using DigitalData.MessagingService.RabbitMQ; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; + +namespace DigitalData.MessagingService.Publisher; + +public static class DependencyInjection +{ + public static IServiceCollection AddMessagingServicePublisher(this IServiceCollection services, Action? configure = null) + { + if(configure is not null) + { + var configuration = new Configuration(services); + configure(configuration); + } + + // --- Email Queue (RabbitMQ) --- + services.AddSingleton(); + + return services; + } + + public class Configuration + { + private readonly IServiceCollection _services; + internal Configuration(IServiceCollection services) + { + _services = services; + } + + public Configuration AddRabbitMqConnectionFactory(IConfiguration configuration) + { + _services.AddRabbitMqConnectionFactory(configuration); + return this; + } + } +} diff --git a/src/infrastructure/DigitalData.MessagingService.Publisher/DigitalData.MessagingService.Publisher.csproj b/src/infrastructure/DigitalData.MessagingService.Publisher/DigitalData.MessagingService.Publisher.csproj index 77a181b..0e91d78 100644 --- a/src/infrastructure/DigitalData.MessagingService.Publisher/DigitalData.MessagingService.Publisher.csproj +++ b/src/infrastructure/DigitalData.MessagingService.Publisher/DigitalData.MessagingService.Publisher.csproj @@ -1,10 +1,19 @@  - net8.0 - enable - enable - latest + net462;net480;net8.0 + enable + enable + latest + + + + + + + + + diff --git a/src/infrastructure/DigitalData.MessagingService.Publisher/OutgoingEmailPublisher.cs b/src/infrastructure/DigitalData.MessagingService.Publisher/OutgoingEmailPublisher.cs new file mode 100644 index 0000000..4192f42 --- /dev/null +++ b/src/infrastructure/DigitalData.MessagingService.Publisher/OutgoingEmailPublisher.cs @@ -0,0 +1,107 @@ +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using DigitalData.MessagingService.RabbitMQ; +using DigitalData.MessagingService.Publisher.Abstraction; + +namespace DigitalData.MessagingService.Publisher; + +/// +/// RabbitMQ-based email queue implementation for outgoing emails. +/// Provides message persistence, scalability, and reliability. +/// Uses Lazy initialization pattern to avoid blocking constructor. +/// +public sealed class OutgoingEmailPublisher : IOutgoingEmailPublisher, IAsyncDisposable +{ + private readonly RabbitMqConfiguration _config; + private readonly ILogger _logger; + private readonly RabbitMqConnectionFactory _cnnFactory; + private readonly Lazy> _lazyChannel; + + public OutgoingEmailPublisher(IOptions config, ILogger logger, RabbitMqConnectionFactory cnnFactory) + { + _config = config.Value; + _logger = logger; + _cnnFactory = cnnFactory; + _lazyChannel = new(InitChannelAsync); + } + + /// + /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. + /// Called lazily on first use via EnsureInitializedAsync. + /// + private async Task InitChannelAsync() + { + var channel = await _cnnFactory.CreateChannelAsync(); + + // Topology declaration can use either channel; use publish channel here + // Declare Dead Letter Queue (DLQ) exchange + await channel.ExchangeDeclareAsync(exchange: _config.DlqExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken); + + // Declare Dead Letter Queue (DLQ) + await channel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: _cnnFactory.CancellationToken); + + // Bind DLQ to DLQ exchange + await channel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: _cnnFactory.CancellationToken); + + // Declare main exchange (Direct type for routing) + await channel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken); + + // Declare main queue (durable for persistence) with DLQ arguments + var queueArgs = new Dictionary + { + { "x-dead-letter-exchange", _config.DlqExchangeName }, + { "x-dead-letter-routing-key", _config.DlqRoutingKey } + }; + + await channel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: _cnnFactory.CancellationToken); + + // Bind main queue to exchange with routing key + await channel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: _cnnFactory.CancellationToken); + + _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName); + + return channel; + } + + public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default) + { + var json = JsonSerializer.Serialize(outgoingEmailEvent); + var body = Encoding.UTF8.GetBytes(json); + + var properties = new BasicProperties + { + Persistent = true, // Message persistence + ContentType = "application/json", + Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) + }; + + var channel = await _lazyChannel.Value; + + await channel.BasicPublishAsync( + exchange: _config.ExchangeName, + routingKey: _config.RoutingKey, + mandatory: false, + basicProperties: properties, + body: body, + cancellationToken: cancellationToken); + } + + public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) + { + var channel = await _lazyChannel.Value; + var queueInfo = await channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); + return (int)queueInfo.MessageCount; + } + + public async ValueTask DisposeAsync() + { + if (await _lazyChannel.Value is IChannel channel) + { + await channel.CloseAsync(); + await channel.DisposeAsync(); + } + } +} diff --git a/src/infrastructure/DigitalData.MessagingService.RabbitMQ/DigitalData.MessagingService.RabbitMQ.csproj b/src/infrastructure/DigitalData.MessagingService.RabbitMQ/DigitalData.MessagingService.RabbitMQ.csproj index 072ed0c..178ab8f 100644 --- a/src/infrastructure/DigitalData.MessagingService.RabbitMQ/DigitalData.MessagingService.RabbitMQ.csproj +++ b/src/infrastructure/DigitalData.MessagingService.RabbitMQ/DigitalData.MessagingService.RabbitMQ.csproj @@ -1,7 +1,7 @@  - net462;net8.0 + net462;net480;net8.0