InMemoryMessageBus.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.Buses;

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

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

    public TransportKind Kind => TransportKind.InMemory;

    public Task ConnectAsync(CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        State = ConnectionState.Connected;
        return Task.CompletedTask;
    }

    public async Task PublishAsync(
        MessageEnvelope message,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(message);
 
        if (State != ConnectionState.Connected)
        {
            throw new InvalidOperationException("The message bus is not connected.");
        }
 
        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)
        {
            cancellationToken.ThrowIfCancellationRequested();
            await handler(message, cancellationToken).ConfigureAwait(false);
        }
    }

    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)
    {
        cancellationToken.ThrowIfCancellationRequested();
        State = ConnectionState.Disconnected;
        return Task.CompletedTask;
    }

    public ValueTask DisposeAsync()
    {
        _handlers.Clear();
        State = ConnectionState.Disconnected;
        return ValueTask.CompletedTask;
    }
}