我尝试从ReadOnlySequence
. 数据被格式化为帧。每个帧都由一个 NULL 字节(八位字节 0)终止。
我的代码使用ReadOnlySequence.PositionOf
. 当它找到一个 NULL 字节时,它将处理直到 NULL 字节位置的所有字节。处理后,我想通过切片输入来处理下一帧并重复前面的步骤。由于帧在 NULL 字节之前结束,因此如果我不再对输入数据进行切片(start = 1),NULL 字节将成为下一个字节序列的一部分。
ReadOnlySequence
是否有一种方法可以用SequencePosition
+ 1 项/字节作为起始值来对 a 进行切片?
我尝试使用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;
}
}