TransportMessageBusAdapter.cs

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

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Dreamine.Communication.Abstractions.Enums;
using Dreamine.Communication.Abstractions.Interfaces;
using Dreamine.Communication.Abstractions.Models;
 
namespace Dreamine.Communication.Core.Adapters;

public sealed class TransportMessageBusAdapter : IMessageBus, IDisposable
{
    private readonly IMessageTransport _transport;
    private readonly ConcurrentDictionary<string, List<Func<MessageEnvelope, CancellationToken, Task>>> _handlers = new();

    public Action<Exception, string>? OnHandlerError { get; set; }

    public TransportMessageBusAdapter(IMessageTransport transport)
    {
        _transport = transport ?? throw new ArgumentNullException(nameof(transport));
        _transport.MessageReceived += OnMessageReceived;
    }

    public ConnectionState State => _transport.State;

    public TransportKind Kind => _transport.Kind;

    public Task ConnectAsync(CancellationToken cancellationToken = default)
    {
        return _transport.ConnectAsync(cancellationToken);
    }

    public Task PublishAsync(
        MessageEnvelope message,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(message);
        return _transport.SendAsync(message, cancellationToken);
    }

    public Task SubscribeAsync(
        string route,
        Func<MessageEnvelope, CancellationToken, Task> handler,
        CancellationToken cancellationToken = default)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(route);
        ArgumentNullException.ThrowIfNull(handler);
        cancellationToken.ThrowIfCancellationRequested();
 
        var handlers = _handlers.GetOrAdd(
            route,
            _ => new List<Func<MessageEnvelope, CancellationToken, Task>>());
 
        lock (handlers)
        {
            handlers.Add(handler);
        }
 
        return Task.CompletedTask;
    }

    public Task DisconnectAsync(CancellationToken cancellationToken = default)
    {
        return _transport.DisconnectAsync(cancellationToken);
    }

    public void Dispose()
    {
        _transport.MessageReceived -= OnMessageReceived;
        _ = _transport.DisposeAsync().AsTask();
    }

    public async ValueTask DisposeAsync()
    {
        _transport.MessageReceived -= OnMessageReceived;
        await _transport.DisposeAsync().ConfigureAwait(false);
    }

    private async void OnMessageReceived(object? sender, MessageEnvelope message)
    {
        try
        {
            if (!_handlers.TryGetValue(message.Route, out var handlers))
            {
                return;
            }
 
            List<Func<MessageEnvelope, CancellationToken, Task>> snapshot;
 
            lock (handlers)
            {
                snapshot = new List<Func<MessageEnvelope, CancellationToken, Task>>(handlers);
            }
 
            foreach (var handler in snapshot)
            {
                await handler(message, CancellationToken.None).ConfigureAwait(false);
            }
        }
        catch (Exception ex)
        {
            // Event dispatchers cannot propagate async-void exceptions.
            // Delegate to caller-supplied handler so failures are observable.
            // 이벤트 디스패처는 async void 예외를 전파할 수 없습니다.
            // 호출자가 OnHandlerError를 설정하면 실패를 관찰할 수 있습니다.
            OnHandlerError?.Invoke(ex, message.Route);
        }
    }
}