DelimiterMessageFrameCodec.cs

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

using System.Runtime.CompilerServices;
using System.Text;
 
namespace Dreamine.Communication.Core.Framing;

public sealed class DelimiterMessageFrameCodec : IMessageFrameCodec
{
    private readonly byte[] _delimiter;
    private readonly int _maxFrameLength;
    private readonly ConditionalWeakTable<Stream, FrameReadBuffer> _readBuffers = new();

    public DelimiterMessageFrameCodec()
        : this("\r\n", Encoding.UTF8, 1024 * 1024)
    {
    }

    public DelimiterMessageFrameCodec(
        string delimiter,
        Encoding encoding,
        int maxFrameLength)
    {
        ArgumentException.ThrowIfNullOrEmpty(delimiter);
        ArgumentNullException.ThrowIfNull(encoding);
 
        if (maxFrameLength <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(maxFrameLength));
        }
 
        _delimiter = encoding.GetBytes(delimiter);
        _maxFrameLength = maxFrameLength;
    }

    public async Task WriteFrameAsync(
        Stream stream,
        byte[] payload,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(stream);
        ArgumentNullException.ThrowIfNull(payload);
 
        await stream.WriteAsync(payload, cancellationToken).ConfigureAwait(false);
        await stream.WriteAsync(_delimiter, cancellationToken).ConfigureAwait(false);
        await stream.FlushAsync(cancellationToken).ConfigureAwait(false);
    }

    public async Task<byte[]?> ReadFrameAsync(
        Stream stream,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(stream);
 
        var frame = new List<byte>(Math.Min(_maxFrameLength, 4096));
        var temp = new byte[Math.Min(_maxFrameLength, 4096)];
        var readBuffer = _readBuffers.GetValue(stream, _ => new FrameReadBuffer());
 
        while (true)
        {
            if (TryConsumeBufferedBytes(readBuffer, frame, out var frameBytes))
            {
                return frameBytes;
            }
 
            var read = await stream.ReadAsync(
                    temp.AsMemory(0, temp.Length),
                    cancellationToken)
                .ConfigureAwait(false);
 
            if (read == 0)
            {
                return frame.Count == 0
                    ? null
                    : frame.ToArray();
            }
 
            lock (readBuffer.SyncRoot)
            {
                for (var index = 0; index < read; index++)
                {
                    readBuffer.Pending.Enqueue(temp[index]);
                }
            }
        }
    }

    private bool TryConsumeBufferedBytes(
        FrameReadBuffer readBuffer,
        List<byte> frame,
        out byte[]? result)
    {
        result = null;
 
        lock (readBuffer.SyncRoot)
        {
            while (readBuffer.Pending.Count > 0)
            {
                frame.Add(readBuffer.Pending.Dequeue());
 
                if (frame.Count > _maxFrameLength)
                {
                    throw new InvalidDataException(
                        $"Frame length exceeded. MaxFrameLength={_maxFrameLength}");
                }
 
                if (!EndsWithDelimiter(frame))
                {
                    continue;
                }
 
                frame.RemoveRange(
                    frame.Count - _delimiter.Length,
                    _delimiter.Length);
 
                result = frame.ToArray();
                return true;
            }
        }
 
        return false;
    }

    private bool EndsWithDelimiter(List<byte> frame)
    {
        if (frame.Count < _delimiter.Length)
        {
            return false;
        }
 
        var start = frame.Count - _delimiter.Length;
 
        for (var index = 0; index < _delimiter.Length; index++)
        {
            if (frame[start + index] != _delimiter[index])
            {
                return false;
            }
        }
 
        return true;
    }

    private sealed class FrameReadBuffer
    {
        public object SyncRoot { get; } = new();

        public Queue<byte> Pending { get; } = [];
    }
}