Add RabbitMQ-based email publisher and DI support
Introduced `OutgoingEmailPublisher` for RabbitMQ-based email queueing with message persistence, scalability, and reliability. Added dependency injection support via `AddMessagingServicePublisher` extension method. Enhanced RabbitMQ topology setup with exchanges, queues, and Dead Letter Queues (DLQ). Updated `DigitalData.MessagingService.Publisher.csproj` and `DigitalData.MessagingService.RabbitMQ.csproj` to support `net462`, `net480`, and `net8.0`. Added project references and conditional package references for compatibility. Integrated logging with `Microsoft.Extensions.Logging` and used `System.Text.Json` for serialization. Implemented lazy initialization for RabbitMQ channels to improve performance.
This commit is contained in:
@@ -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<Configuration>? configure = null)
|
||||||
|
{
|
||||||
|
if(configure is not null)
|
||||||
|
{
|
||||||
|
var configuration = new Configuration(services);
|
||||||
|
configure(configuration);
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Email Queue (RabbitMQ) ---
|
||||||
|
services.AddSingleton<IOutgoingEmailPublisher, OutgoingEmailPublisher>();
|
||||||
|
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,10 +1,19 @@
|
|||||||
<Project Sdk="Microsoft.NET.Sdk">
|
<Project Sdk="Microsoft.NET.Sdk">
|
||||||
|
|
||||||
<PropertyGroup>
|
<PropertyGroup>
|
||||||
<TargetFramework>net8.0</TargetFramework>
|
<TargetFrameworks>net462;net480;net8.0</TargetFrameworks>
|
||||||
<ImplicitUsings>enable</ImplicitUsings>
|
<ImplicitUsings>enable</ImplicitUsings>
|
||||||
<Nullable>enable</Nullable>
|
<Nullable>enable</Nullable>
|
||||||
<LangVersion>latest</LangVersion>
|
<LangVersion>latest</LangVersion>
|
||||||
</PropertyGroup>
|
</PropertyGroup>
|
||||||
|
|
||||||
|
<ItemGroup Condition="'$(TargetFrameworkIdentifier)' == '.NETFramework'">
|
||||||
|
<PackageReference Include="System.Text.Json" Version="10.0.10" />
|
||||||
|
</ItemGroup>
|
||||||
|
|
||||||
|
<ItemGroup>
|
||||||
|
<ProjectReference Include="..\..\core\DigitalData.MessagingService.Publisher.Abstraction\DigitalData.MessagingService.Publisher.Abstraction.csproj" />
|
||||||
|
<ProjectReference Include="..\DigitalData.MessagingService.RabbitMQ\DigitalData.MessagingService.RabbitMQ.csproj" />
|
||||||
|
</ItemGroup>
|
||||||
|
|
||||||
</Project>
|
</Project>
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// RabbitMQ-based email queue implementation for outgoing emails.
|
||||||
|
/// Provides message persistence, scalability, and reliability.
|
||||||
|
/// Uses Lazy<T> initialization pattern to avoid blocking constructor.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class OutgoingEmailPublisher : IOutgoingEmailPublisher, IAsyncDisposable
|
||||||
|
{
|
||||||
|
private readonly RabbitMqConfiguration _config;
|
||||||
|
private readonly ILogger<OutgoingEmailPublisher> _logger;
|
||||||
|
private readonly RabbitMqConnectionFactory _cnnFactory;
|
||||||
|
private readonly Lazy<Task<IChannel>> _lazyChannel;
|
||||||
|
|
||||||
|
public OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailPublisher> logger, RabbitMqConnectionFactory cnnFactory)
|
||||||
|
{
|
||||||
|
_config = config.Value;
|
||||||
|
_logger = logger;
|
||||||
|
_cnnFactory = cnnFactory;
|
||||||
|
_lazyChannel = new(InitChannelAsync);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
|
||||||
|
/// Called lazily on first use via EnsureInitializedAsync.
|
||||||
|
/// </summary>
|
||||||
|
private async Task<IChannel> 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<string, object?>
|
||||||
|
{
|
||||||
|
{ "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<int> 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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
<Project Sdk="Microsoft.NET.Sdk">
|
<Project Sdk="Microsoft.NET.Sdk">
|
||||||
|
|
||||||
<PropertyGroup>
|
<PropertyGroup>
|
||||||
<TargetFrameworks>net462;net8.0</TargetFrameworks>
|
<TargetFrameworks>net462;net480;net8.0</TargetFrameworks>
|
||||||
</PropertyGroup>
|
</PropertyGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
|
|||||||
Reference in New Issue
Block a user