RabbitMqMessageBus.cs

이 파일의 문서화 페이지로 가기

using System.Text.Json;
using Dreamine.Communication.Abstractions.Enums;
using Dreamine.Communication.Abstractions.Interfaces;
using Dreamine.Communication.Abstractions.Models;
using Dreamine.Communication.RabbitMQ.Infrastructure;
using Dreamine.Communication.RabbitMQ.Options;
 
namespace Dreamine.Communication.RabbitMQ.Buses;

public sealed class RabbitMqMessageBus : IMessageBus
{
    private readonly RabbitMqMessageBusOptions _options;
    private readonly IRabbitMqConnectionFactory _connectionFactory;
    private readonly Dictionary<string, List<Func<MessageEnvelope, CancellationToken, Task>>> _handlers = new();
    private readonly object _syncRoot = new();

    private IRabbitMqConnection? _connection;
    private IRabbitMqChannel? _channel;
    private string? _consumerTag;

    public RabbitMqMessageBus(RabbitMqMessageBusOptions options)
        : this(options, new RabbitMqClientConnectionFactory())
    {
    }

    public RabbitMqMessageBus(
        RabbitMqMessageBusOptions options,
        IRabbitMqConnectionFactory connectionFactory)
    {
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory));
        ValidateOptions(_options);
    }

    public ConnectionState State { get; private set; } = ConnectionState.Disconnected;

    public TransportKind Kind => TransportKind.RabbitMq;

    public Task ConnectAsync(CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
 
        if (State == ConnectionState.Connected)
        {
            return Task.CompletedTask;
        }
 
        State = ConnectionState.Connecting;
 
        try
        {
            _connection = _connectionFactory.CreateConnection(_options);
            _channel = _connection.CreateChannel();
 
            DeclareTopology(_channel);
 
            State = ConnectionState.Connected;
            return Task.CompletedTask;
        }
        catch
        {
            State = ConnectionState.Faulted;
            Cleanup();
            throw;
        }
    }

    public Task PublishAsync(
        MessageEnvelope message,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(message);
        cancellationToken.ThrowIfCancellationRequested();
 
        if (_channel is null || State != ConnectionState.Connected)
        {
            throw new InvalidOperationException("RabbitMQ message bus is not connected.");
        }
 
        var route = string.IsNullOrWhiteSpace(message.Route)
            ? _options.RoutingKey
            : message.Route;
 
        var body = JsonSerializer.SerializeToUtf8Bytes(message);
 
        var properties = _channel.CreateBasicProperties();
        properties.Persistent = _options.PersistentMessages;
        properties.ContentType = "application/json";
        properties.Type = nameof(MessageEnvelope);
 
        _channel.BasicPublish(
            exchange: _options.ExchangeName,
            routingKey: route,
            mandatory: false,
            properties: properties,
            body: body);
 
        return Task.CompletedTask;
    }

    public Task SubscribeAsync(
        string route,
        Func<MessageEnvelope, CancellationToken, Task> handler,
        CancellationToken cancellationToken = default)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(route);
        ArgumentNullException.ThrowIfNull(handler);
        cancellationToken.ThrowIfCancellationRequested();
 
        if (_channel is null || State != ConnectionState.Connected)
        {
            throw new InvalidOperationException("RabbitMQ message bus is not connected.");
        }
 
        lock (_syncRoot)
        {
            if (!_handlers.TryGetValue(route, out var handlers))
            {
                handlers = [];
                _handlers[route] = handlers;
            }
 
            handlers.Add(handler);
        }
 
        _channel.QueueBind(
            queue: _options.QueueName,
            exchange: _options.ExchangeName,
            routingKey: route);
 
        if (!string.IsNullOrWhiteSpace(_consumerTag))
        {
            return Task.CompletedTask;
        }
 
        _consumerTag = _channel.BasicConsume(
            queue: _options.QueueName,
            autoAck: false,
            onReceived: (delivery, token) => HandleReceivedAsync(
                delivery,
                token.CanBeCanceled ? token : cancellationToken));
 
        return Task.CompletedTask;
    }

    public Task DisconnectAsync(CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
 
        State = ConnectionState.Disconnecting;
 
        try
        {
            Cleanup();
            State = ConnectionState.Disconnected;
            return Task.CompletedTask;
        }
        catch
        {
            State = ConnectionState.Faulted;
            throw;
        }
    }

    public async ValueTask DisposeAsync()
    {
        await DisconnectAsync().ConfigureAwait(false);
    }

    private async Task HandleReceivedAsync(
        RabbitMqDelivery delivery,
        CancellationToken cancellationToken)
    {
        if (_channel is null)
        {
            return;
        }
 
        try
        {
            var message = JsonSerializer.Deserialize<MessageEnvelope>(delivery.Body.Span);
 
            if (message is null)
            {
                _channel.BasicReject(delivery.DeliveryTag, requeue: false);
                return;
            }
 
            var route = string.IsNullOrWhiteSpace(message.Route)
                ? delivery.RoutingKey
                : message.Route;
 
            List<Func<MessageEnvelope, CancellationToken, Task>> handlers;
 
            lock (_syncRoot)
            {
                if (!_handlers.TryGetValue(route, out var routeHandlers) &&
                    !_handlers.TryGetValue(delivery.RoutingKey, out routeHandlers))
                {
                    handlers = [];
                }
                else
                {
                    handlers = new List<Func<MessageEnvelope, CancellationToken, Task>>(routeHandlers);
                }
            }
 
            if (handlers.Count == 0)
            {
                _channel.BasicAck(delivery.DeliveryTag, multiple: false);
                return;
            }
 
            foreach (var handler in handlers)
            {
                await handler(message, cancellationToken).ConfigureAwait(false);
            }
 
            _channel.BasicAck(delivery.DeliveryTag, multiple: false);
        }
        catch
        {
            try
            {
                _channel.BasicNack(
                    delivery.DeliveryTag,
                    multiple: false,
                    requeue: _options.RequeueOnHandlerError);
            }
            catch
            {
                // Ignore acknowledgement failure during shutdown or channel fault.
            }
 
            throw;
        }
    }

    private void DeclareTopology(IRabbitMqChannel channel)
    {
        channel.ExchangeDeclare(
            exchange: _options.ExchangeName,
            type: _options.ExchangeType,
            durable: _options.Durable,
            autoDelete: _options.AutoDelete);
 
        channel.QueueDeclare(
            queue: _options.QueueName,
            durable: _options.Durable,
            exclusive: _options.Exclusive,
            autoDelete: _options.AutoDelete);
 
        channel.QueueBind(
            queue: _options.QueueName,
            exchange: _options.ExchangeName,
            routingKey: _options.RoutingKey);
    }

    private void Cleanup()
    {
        lock (_syncRoot)
        {
            _handlers.Clear();
        }
 
        if (_channel is not null)
        {
            try
            {
                if (!string.IsNullOrWhiteSpace(_consumerTag))
                {
                    _channel.BasicCancel(_consumerTag);
                }
            }
            catch
            {
                // Ignore consumer cancel failure during cleanup.
            }
 
            try
            {
                if (_channel.IsOpen)
                {
                    _channel.Close();
                }
            }
            catch
            {
                // Ignore channel close failure during cleanup.
            }
 
            _channel.Dispose();
            _channel = null;
        }
 
        if (_connection is not null)
        {
            try
            {
                if (_connection.IsOpen)
                {
                    _connection.Close();
                }
            }
            catch
            {
                // Ignore connection close failure during cleanup.
            }
 
            _connection.Dispose();
            _connection = null;
        }
 
        _consumerTag = null;
    }

    private static void ValidateOptions(RabbitMqMessageBusOptions options)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(options.HostName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.VirtualHost);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.UserName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.Password);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.ExchangeName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.QueueName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.RoutingKey);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.ExchangeType);
 
        if (options.Port <= 0 || options.Port > 65535)
        {
            throw new ArgumentOutOfRangeException(nameof(options.Port));
        }
    }
}