c# - ReadOnlySequence – 切片到给定的 SequencePosition + 1

Question

我尝试从 ReadOnlySequence 中读取一些数据。数据被格式化为帧。每个帧都以 NULL 字节(八位字节 0)结束。

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

有没有办法用 SequencePosition + 1 个项目/字节作为起始值来切片 ReadOnlySequence？

我尝试使用 SequencePosition.GetInteger + 1 作为起始值，但这不起作用，因为 GetInteger 有时返回的值大于 ReadOnlySequence 的长度。对 GetInteger 返回的值进行切片会导致以下异常:System.ArgumentOutOfRangeException:指定的参数超出了有效值的范围。 (参数'开始')

最小可重现示例

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 之外会发生什么，但我想您在尝试对缓冲区进行切片时会遇到问题。但是，在您的特定情况下，您已经知道该字节在那里。