src/Shared/NotificationService.Shared/Infrastructure/RabbitMQMessageQueue.cs - master - VFlov/NotificationService_1

src/Shared/NotificationService.Shared/Infrastructure/RabbitMQMessageQueue.cs
241 строка8 KB
VFlov
Update
25 дек 2025, 21:02
25 дек 2025, 21:02cf3965d
using Microsoft.Extensions.Logging;
using NotificationService.Shared.Interfaces;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;

namespace NotificationService.Shared.Infrastructure;

/// <summary>
/// Реализация очереди сообщений на базе RabbitMQ
/// </summary>
public class RabbitMQMessageQueue : IMessageQueue, IDisposable
{
    private readonly IConnection _connection;
    private readonly IChannel _channel;
    private readonly ILogger<RabbitMQMessageQueue> _logger;
    private readonly JsonSerializerOptions _jsonOptions;

    public RabbitMQMessageQueue(string connectionString, ILogger<RabbitMQMessageQueue> logger)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        
        if (string.IsNullOrWhiteSpace(connectionString))
        {
            throw new ArgumentException("Connection string cannot be empty", nameof(connectionString));
        }

        _jsonOptions = new JsonSerializerOptions
        {
            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
            WriteIndented = false
        };

        try
        {
            _logger.LogInformation("Connecting to RabbitMQ: {ConnectionString}", MaskConnectionString(connectionString));
            
            var factory = new ConnectionFactory { Uri = new Uri(connectionString) };
            _connection = factory.CreateConnectionAsync().GetAwaiter().GetResult();
            _channel = _connection.CreateChannelAsync().GetAwaiter().GetResult();
            
            _logger.LogInformation("RabbitMQ connection established successfully");
        }
        catch (Exception ex)
        {
            _logger.LogCritical(ex, "Failed to establish RabbitMQ connection");
            throw;
        }
    }

    public async Task PublishAsync<T>(string queueName, T message, CancellationToken cancellationToken = default) where T : class
    {
        if (string.IsNullOrWhiteSpace(queueName))
        {
            throw new ArgumentException("Queue name cannot be empty", nameof(queueName));
        }

        if (message == null)
        {
            throw new ArgumentNullException(nameof(message));
        }

        try
        {
            _logger.LogDebug("Declaring queue: {QueueName}", queueName);
            
            await _channel.QueueDeclareAsync(
                queue: queueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: null,
                cancellationToken: cancellationToken);

            var json = JsonSerializer.Serialize(message, _jsonOptions);
            var body = Encoding.UTF8.GetBytes(json);

            _logger.LogDebug(
                "Publishing message to queue: {QueueName}, MessageSize: {Size} bytes",
                queueName, body.Length);

            var properties = new BasicProperties
            {
                Persistent = true,
                DeliveryMode = DeliveryModes.Persistent
            };

            await _channel.BasicPublishAsync(
                exchange: string.Empty,
                routingKey: queueName,
                mandatory: false,
                basicProperties: properties,
                body: body,
                cancellationToken: cancellationToken);

            _logger.LogInformation("Message published successfully to queue: {QueueName}", queueName);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, 
                "Error publishing message to queue: {QueueName}",
                queueName);
            throw;
        }
    }

    public async Task SubscribeAsync<T>(string queueName, Func<T, Task<bool>> handler, CancellationToken cancellationToken = default) where T : class
    {
        if (string.IsNullOrWhiteSpace(queueName))
        {
            throw new ArgumentException("Queue name cannot be empty", nameof(queueName));
        }

        if (handler == null)
        {
            throw new ArgumentNullException(nameof(handler));
        }

        try
        {
            _logger.LogInformation("Subscribing to queue: {QueueName}", queueName);
            
            await _channel.QueueDeclareAsync(
                queue: queueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: null,
                cancellationToken: cancellationToken);

            await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken);

            var consumer = new AsyncEventingBasicConsumer(_channel);
            
            consumer.ReceivedAsync += async (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var json = Encoding.UTF8.GetString(body);

                try
                {
                    _logger.LogDebug(
                        "Received message from queue: {QueueName}, DeliveryTag: {DeliveryTag}",
                        queueName, ea.DeliveryTag);

                    var message = JsonSerializer.Deserialize<T>(json, _jsonOptions);
                    if (message == null)
                    {
                        _logger.LogWarning(
                            "Failed to deserialize message from queue: {QueueName}",
                            queueName);
                        await _channel.BasicNackAsync(ea.DeliveryTag, false, false);
                        return;
                    }

                    var success = await handler(message);
                    
                    if (success)
                    {
                        await _channel.BasicAckAsync(ea.DeliveryTag, false);
                        _logger.LogDebug(
                            "Message acknowledged successfully - Queue: {QueueName}, DeliveryTag: {DeliveryTag}",
                            queueName, ea.DeliveryTag);
                    }
                    else
                    {
                        // Negative acknowledgment with requeue
                        await _channel.BasicNackAsync(ea.DeliveryTag, false, true);
                        _logger.LogWarning(
                            "Message processing failed, requeued - Queue: {QueueName}, DeliveryTag: {DeliveryTag}",
                            queueName, ea.DeliveryTag);
                    }
                }
                catch (JsonException ex)
                {
                    _logger.LogError(ex,
                        "JSON deserialization error for message from queue: {QueueName}",
                        queueName);
                    // Don't requeue invalid messages
                    await _channel.BasicNackAsync(ea.DeliveryTag, false, false);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex,
                        "Error processing message from queue: {QueueName}, DeliveryTag: {DeliveryTag}",
                        queueName, ea.DeliveryTag);
                    // Don't requeue on handler error (already handled by worker)
                    await _channel.BasicNackAsync(ea.DeliveryTag, false, false);
                }
            };

            await _channel.BasicConsumeAsync(
                queue: queueName,
                autoAck: false,
                consumer: consumer,
                cancellationToken: cancellationToken);

            _logger.LogInformation("Successfully subscribed to queue: {QueueName}", queueName);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error subscribing to queue: {QueueName}", queueName);
            throw;
        }
    }

    public void Dispose()
    {
        try
        {
            _channel?.Dispose();
            _connection?.Dispose();
            _logger.LogInformation("RabbitMQ connection disposed");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error disposing RabbitMQ connection");
        }
    }

    /// <summary>
    /// Маскирует пароль в строке подключения для безопасного логирования
    /// </summary>
    private static string MaskConnectionString(string connectionString)
    {
        try
        {
            var uri = new Uri(connectionString);
            if (!string.IsNullOrEmpty(uri.UserInfo))
            {
                return connectionString.Replace(uri.UserInfo, "****:****");
            }
            return connectionString;
        }
        catch
        {
            return "****";
        }
    }
}