Add RabbitMQ support with configuration and connection
Added support for RabbitMQ integration: - Updated project to target `net462` and `net8.0`. - Added NuGet dependencies: `RabbitMQ.Client`, `Microsoft.Extensions.Logging.Abstractions`, and `Microsoft.Extensions.Options.ConfigurationExtensions`. - Introduced `RabbitMqConfiguration` class for managing RabbitMQ settings. - Implemented `RabbitMqConnectionFactory` for creating and managing RabbitMQ connections with lazy initialization, async disposal, and logging support.
This commit is contained in:
@@ -5,7 +5,9 @@
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.10" />
|
||||
<PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
|
||||
<PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.10" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
namespace DigitalData.MessagingService.RabbitMQ
|
||||
{
|
||||
/// <summary>
|
||||
/// Configuration for RabbitMQ connection
|
||||
/// </summary>
|
||||
public class RabbitMqConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Configuration section name in appsettings.json
|
||||
/// </summary>
|
||||
public const string SectionName = "RabbitMQ";
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ server hostname
|
||||
/// </summary>
|
||||
public string HostName { get; set; } = "localhost";
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ AMQP port (default: 5672)
|
||||
/// </summary>
|
||||
public int Port { get; set; } = 5672;
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ username
|
||||
/// </summary>
|
||||
public string UserName { get; set; } = "guest";
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ password
|
||||
/// </summary>
|
||||
public string Password { get; set; } = "guest";
|
||||
|
||||
/// <summary>
|
||||
/// Virtual host (default: /)
|
||||
/// </summary>
|
||||
public string VirtualHost { get; set; } = "/";
|
||||
|
||||
/// <summary>
|
||||
/// Enable automatic recovery on connection failure
|
||||
/// </summary>
|
||||
public bool AutomaticRecoveryEnabled { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// Network recovery interval in seconds
|
||||
/// </summary>
|
||||
public int NetworkRecoveryIntervalSeconds { get; set; } = 10;
|
||||
|
||||
public string QueueName { get; set; }
|
||||
public string ExchangeName { get; set; }
|
||||
public string RoutingKey { get; set; }
|
||||
public string DlqQueueName { get; set; }
|
||||
public string DlqExchangeName { get; set; }
|
||||
public string DlqRoutingKey { get; set; }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using RabbitMQ.Client;
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace DigitalData.MessagingService.RabbitMQ
|
||||
{
|
||||
public sealed class RabbitMqConnectionFactory : IAsyncDisposable
|
||||
{
|
||||
private readonly RabbitMqConfiguration _config;
|
||||
|
||||
private readonly ILogger<RabbitMqConnectionFactory>
|
||||
#if nullable
|
||||
?
|
||||
#endif
|
||||
_logger;
|
||||
|
||||
private readonly Lazy<Task<IConnection>> _lazyConnectionProvider;
|
||||
|
||||
private IConnection
|
||||
#if nullable
|
||||
?
|
||||
#endif
|
||||
_connection = null;
|
||||
|
||||
private CancellationToken? _cancellationToken;
|
||||
|
||||
public CancellationToken CancellationToken => _cancellationToken
|
||||
?? throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method.");
|
||||
|
||||
public Task<IConnection> GetConnectionAsync()
|
||||
{
|
||||
if (_cancellationToken != null)
|
||||
return _lazyConnectionProvider.Value;
|
||||
else
|
||||
throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method.");
|
||||
}
|
||||
|
||||
public RabbitMqConnectionFactory(IOptions<RabbitMqConfiguration> config)
|
||||
{
|
||||
_config = config.Value;
|
||||
_lazyConnectionProvider = new Lazy<Task<IConnection>>(async () =>
|
||||
{
|
||||
_cancellationToken?.ThrowIfCancellationRequested();
|
||||
var factory = new ConnectionFactory
|
||||
{
|
||||
HostName = _config.HostName,
|
||||
Port = _config.Port,
|
||||
UserName = _config.UserName,
|
||||
Password = _config.Password,
|
||||
VirtualHost = _config.VirtualHost,
|
||||
AutomaticRecoveryEnabled = _config.AutomaticRecoveryEnabled,
|
||||
NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds),
|
||||
};
|
||||
|
||||
_connection = await factory.CreateConnectionAsync((CancellationToken)_cancellationToken
|
||||
#if nullable
|
||||
!
|
||||
#endif
|
||||
);
|
||||
return _connection;
|
||||
});
|
||||
}
|
||||
|
||||
public async Task InitAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
_cancellationToken = cancellationToken;
|
||||
_ = await GetConnectionAsync();
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_connection != null)
|
||||
{
|
||||
await _connection.CloseAsync();
|
||||
await _connection.DisposeAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user