InMemoryOutboundMessageQueue.cs

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

using Dreamine.Communication.Abstractions.Options;
 
namespace Dreamine.Communication.Core.Queues;

public sealed class InMemoryOutboundMessageQueue : IOutboundMessageQueue
{
    private readonly object _sync = new();
    private readonly LinkedList<QueuedOutboundMessage> _messages = new();
    private readonly OutboundQueueOptions _options;

    public InMemoryOutboundMessageQueue(OutboundQueueOptions? options = null)
    {
        _options = options ?? new OutboundQueueOptions();
 
        if (_options.MaxQueueSize <= 0)
        {
            throw new ArgumentOutOfRangeException(
                nameof(options),
                "MaxQueueSize must be greater than zero.");
        }
    }

    public int Count
    {
        get
        {
            lock (_sync)
            {
                return _messages.Count;
            }
        }
    }

    public ValueTask EnqueueAsync(
        QueuedOutboundMessage message,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(message);
        cancellationToken.ThrowIfCancellationRequested();
 
        lock (_sync)
        {
            EnsureCapacityForAddToBack();
            _messages.AddLast(message);
        }
 
        return ValueTask.CompletedTask;
    }

    public ValueTask EnqueueFrontAsync(
        QueuedOutboundMessage message,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(message);
        cancellationToken.ThrowIfCancellationRequested();
 
        lock (_sync)
        {
            EnsureCapacityForAddToFront();
            _messages.AddFirst(message);
        }
 
        return ValueTask.CompletedTask;
    }

    public ValueTask<QueuedOutboundMessage?> TryDequeueAsync(
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
 
        lock (_sync)
        {
            if (_messages.First is null)
            {
                return ValueTask.FromResult<QueuedOutboundMessage?>(null);
            }
 
            var message = _messages.First.Value;
            _messages.RemoveFirst();
 
            return ValueTask.FromResult<QueuedOutboundMessage?>(message);
        }
    }

    public void Clear()
    {
        lock (_sync)
        {
            _messages.Clear();
        }
    }

    private void EnsureCapacityForAddToBack()
    {
        if (_messages.Count < _options.MaxQueueSize)
        {
            return;
        }
 
        if (!_options.DropOldestWhenFull)
        {
            throw new InvalidOperationException("Outbound message queue is full.");
        }
 
        _messages.RemoveFirst();
    }

    private void EnsureCapacityForAddToFront()
    {
        if (_messages.Count < _options.MaxQueueSize)
        {
            return;
        }
 
        if (!_options.DropOldestWhenFull)
        {
            throw new InvalidOperationException("Outbound message queue is full.");
        }
 
        _messages.RemoveLast();
    }
}