ReadOnlySequence – 切片到给定的 SequencePosition + 1

Question

我尝试从ReadOnlySequence 读取一些数据。数据被格式化为帧。每个帧都以 NULL 字节（八位字节 0）终止。

我的代码使用ReadOnlySequence.PositionOf 搜索帧的结尾。当它找到一个 NULL 字节时，它将处理直到 NULL 字节位置的所有字节。处理后，我想通过切片输入来处理下一帧并重复前面的步骤。 由于帧在 NULL 字节之前结束，如果我不再对输入数据进行切片（start = 1），NULL 字节将成为下一个字节序列的一部分。

是否有一种方法可以将 ReadOnlySequence 与 SequencePosition + 1 项/字节作为起始值进行切片？

我尝试使用SequencePosition.GetInteger + 1 作为起始值，但这不起作用，因为GetInteger 有时返回的值大于ReadOnlySequence 的长度。切片到GetInteger 返回的值会导致以下异常：System.ArgumentOutOfRangeException: Specified argument was out of the range of valid values. (Parameter 'start')

最小可重现示例

using System;
using System.Buffers;
using System.IO.Pipelines;
using System.Text;
using System.Threading.Tasks;

namespace ConsoleApp1
{
    public class Program
    {
        private static IDuplexPipe _pipe;

        public static async Task Main( String[] args )
        {
            var pipe = new Pipe();
            _pipe = new DuplexPipe( pipe.Reader, pipe.Writer );

            var firstMessage = Encoding.UTF8.GetBytes( "CONNECTED\nversion:1.1\nsession:2a840965\nserver:ActiveMQ-Artemis/2.8.0 ActiveMQ Artemis Messaging Engine\nheart-beat:10000,10000\n\n\0\n" );
            await _pipe.Output.WriteAsync( firstMessage );
            await _pipe.Output.FlushAsync();

            var secondMessage =
                Encoding.UTF8.GetBytes(
                    "\n\nMESSAGE\nsubscription:test-839c7766-0f38-4579-a3fc-74de35408536Sub1\ncontent-length:4\nmessage-id:2147486350\ndestination:/queue/TestQ\nexpires:1572278642017\nredelivered:false\npriority:5\npersistent:true\ntimestamp:1572278582050\ndestination-type:ANYCAST\nreceipt:2\ntest:test\nNMSXDeliveryMode:true\ntransformation:jms-byte\ntimestamp:1572278582017\n\nHello World\0\n" );
            await _pipe.Output.WriteAsync( secondMessage );
            await _pipe.Output.FlushAsync();

            var readResult = await _pipe.Input.ReadAsync();
            var buffer = readResult.Buffer;
            while ( TryParseFrame( ref buffer ) )
            {
                // ...
            }

            _pipe.Input.AdvanceTo( buffer.Start, buffer.End );

            Console.ReadLine();
        }

        private static Boolean TryParseFrame( ref ReadOnlySequence<Byte> inputBuffer )
        {
            var endOfFrame = inputBuffer.PositionOf( ByteConstants.Null );
            if ( endOfFrame == null )
                return false;

            var frameBuffer = inputBuffer.Slice( 0, endOfFrame.Value );
            // parse and process the frame...

            // This works....
            //inputBuffer = inputBuffer.Slice( frameBuffer.End );
            //inputBuffer = inputBuffer.Slice( 1 );

            // This does NOT.
            try
            {
                var end = frameBuffer.End.GetInteger();
                var length = inputBuffer.Length;
                Console.WriteLine( $" END: {end}, LENGTH: {length} " );
                inputBuffer = inputBuffer.Slice( end + 1 );
            }
            catch ( Exception ex )
            {
                Console.WriteLine( ex );
                // Make sure we can read the next frame...
                inputBuffer = inputBuffer.Slice( frameBuffer.End );
                inputBuffer = inputBuffer.Slice( 1 );
            }

            return true;
        }
    }

    public class DuplexPipe : IDuplexPipe
    {
        public DuplexPipe( PipeReader input, PipeWriter output )
        {
            Input = input;
            Output = output;
        }

        public PipeReader Input { get; }
        public PipeWriter Output { get; }
    }

    public static class ByteConstants
    {
        public const Byte HeaderDelimiter = 58;
        public const Byte LineFeed = 10;
        public const Byte Null = 0;
    }
}

Answer 1

我最近玩了管道游戏，很快就放弃了操作SequencePosition 构造。

ReadOnlySequence 公开了一个GetPosition(Int64, SequencePosition) 方法，其中第一个参数是从提供的SequencePosition 开始的偏移量（阅读多远）（在您的情况下，为终止空字节返回的位置）。

考虑以下几点：

private static bool TryParseFrame(ref ReadOnlySequence<byte> inputBuffer)
{
    var endOfFrame = inputBuffer.PositionOf(ByteConstants.Null);
    if (endOfFrame == null)
        return false;

    // Get SequencePosition 1 place after endOfFrame
    var sliceEnd = inputBuffer.GetPosition(1, endOfFrame.Value);

    inputBuffer = inputBuffer.Slice(0, sliceEnd);

    return true;
}

不确定以上内容是否正是您要查找的内容，但希望 GetPosition() 能帮到您。

免责声明：我不知道如果您的偏移量超出ReadOnlySequence 会发生什么，但我想您在尝试分割缓冲区时会遇到问题。但是，在您的特定情况下，您已经知道该字节在那里。