diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs index b191789..355fb17 100644 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs @@ -2,7 +2,7 @@ using DigitalData.MessagingService.Application.Common.Interfaces; using DigitalData.MessagingService.Infrastructure.Queue; using DigitalData.MessagingService.Infrastructure.Services; using DigitalData.MessagingService.Infrastructure.Services.Background; -using DigitalData.MessagingService.Publisher.Abstraction; +using DigitalData.MessagingService.Publisher; using DigitalData.MessagingService.RabbitMQ; using Microsoft.AspNetCore.DataProtection; using Microsoft.Extensions.Configuration; @@ -34,7 +34,7 @@ public static class DependencyInjection // --- Email Queue (RabbitMQ) --- services.AddSingleton(); - services.AddSingleton(); + services.AddMessagingServicePublisher(); // --- RabbitMQ Configuration --- services.AddRabbitMqConnectionFactory(configuration); diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj index 7b6305a..b3ec186 100644 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj @@ -9,6 +9,7 @@ + diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs deleted file mode 100644 index 2ff0fa7..0000000 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs +++ /dev/null @@ -1,107 +0,0 @@ -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.Infrastructure.Queue; - -/// -/// 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(); - } - } -}