基于长度字段的System.IO.Pipelines解码

Question

我正在尝试将代码从DotNetty移植到System.IO.Pipelines。在DotNetty中，我利用LengthFieldBasedFrameDecoder对TCP消息进行解码，其中前两个字节表示一个整数，表示整个消息的长度。

我见过的所有演示都依赖于基于字符串的EOL指示器。我觉得这应该很容易，但我不知道如何抓取前两个字节，然后是X个字节，如长度前缀所示。

下面是David Fowler的TcpEcho server中的一个例子。如果前两个字节指示消息大小，而不是指示消息结束的EOL字符，我如何重写它来解析消息？

 private static async Task ReadPipeAsync(Socket socket, PipeReader reader)
    {
        while (true)
        {
            ReadResult result = await reader.ReadAsync();

            ReadOnlySequence<byte> buffer = result.Buffer;
            SequencePosition? position = null;

            do
            {
                // Find the EOL
                position = buffer.PositionOf((byte)'\n');

                if (position != null)
                {
                    var line = buffer.Slice(0, position.Value);
                    ProcessLine(socket, line);

                    // This is equivalent to position + 1
                    var next = buffer.GetPosition(1, position.Value);

                    // Skip what we've already processed including \n
                    buffer = buffer.Slice(next);
                }
            }
            while (position != null);

            // We sliced the buffer until no more data could be processed
            // Tell the PipeReader how much we consumed and how much we left to process
            reader.AdvanceTo(buffer.Start, buffer.End);

            if (result.IsCompleted)
            {
                break;
            }
        }

        reader.Complete();
    }

Answer 1

This是我最终得到的结果：

private const int lengthPrefixSize = 2; // number of bytes in the length prefix
private static ushort ParseLengthPrefix(ReadOnlySpan<byte> buffer) => BinaryPrimitives.ReadUInt16LittleEndian(buffer);

private static ushort ParseLengthPrefix(in ReadOnlySequence<byte> buffer)
{
    if (buffer.First.Length >= lengthPrefixSize)
        return ParseLengthPrefix(buffer.First.Span.Slice(0, lengthPrefixSize));

    Span<byte> lengthPrefixBytes = stackalloc byte[lengthPrefixSize];
    buffer.Slice(0, lengthPrefixSize).CopyTo(lengthPrefixBytes);
    return ParseLengthPrefix(lengthPrefixBytes);
}

private static async Task ReadPipeAsync(Socket socket, PipeReader reader)
{
    ushort? lengthPrefix = null;

    while (true)
    {
        ReadResult result = await reader.ReadAsync();

        ReadOnlySequence<byte> buffer = result.Buffer;

        while (true)
        {
            if (lengthPrefix == null)
            {
                // If we don't have enough for the length prefix, then wait for more data.
                if (buffer.Length < lengthPrefixSize)
                    break;

                // Read and parse the length prefix
                lengthPrefix = ParseLengthPrefix(buffer);
                buffer = buffer.Slice(lengthPrefixSize);
            }

            // If we haven't read the entire packet yet, then wait.
            if (buffer.Length < lengthPrefix.Value)
                break;

            // Read the data packet
            var line = buffer.Slice(0, lengthPrefix.Value);
            ProcessLine(socket, line);

            buffer = buffer.Slice(lengthPrefix.Value);
            lengthPrefix = null;
        }

        // We sliced the buffer until no more data could be processed
        // Tell the PipeReader how much we consumed and how much we left to process
        reader.AdvanceTo(buffer.Start, buffer.End);

        if (result.IsCompleted)
        {
            break;
        }
    }

    reader.Complete();
}

这个解决方案确实有一个长度前缀缓冲区，但它仅在长度前缀跨多个跨度拆分时使用。我认为有一个SequenceReader coming可以让这一切变得完全无拷贝，尽管在使用长度前缀的情况下(非常少的字节和没有缓冲区分配)，差异可能是最小的。