SpanReader支持从数据流中持续读取数据来扩容读写器，以支持redis/mysql等部分数据帧过大的协议。

Author: nnhy

NewLife.Core/Buffers/SpanReader.cs

@@ -7,10 +7,14 @@
 namespace NewLife.Buffers;
 
 /// <summary>Span读取器</summary>
+/// <remarks>
+/// 引用结构的Span读取器确保高性能无GC读取。
+/// 支持Stream扩展，当数据不足时，自动从数据流中读取，常用于解析Redis/MySql等协议。
+/// </remarks>
 public ref struct SpanReader
 {
     #region 属性
-    private readonly ReadOnlySpan<Byte> _span;
+    private ReadOnlySpan<Byte> _span;
     /// <summary>数据片段</summary>
     public ReadOnlySpan<Byte> Span => _span;
 
@@ -39,7 +43,43 @@ public ref struct SpanReader
 
     /// <summary>实例化Span读取器</summary>
     /// <param name="data"></param>
-    public SpanReader(IPacket data) : this(data.GetSpan()) { }
+    public SpanReader(IPacket data)
+    {
+        _data = data;
+        _span = data.GetSpan();
+        _total = data.Total;
+    }
+    #endregion
+
+    #region 扩容增强
+    /// <summary>最大容量。多次从数据流读取数据时，受限于此最大值</summary>
+    public Int32 MaxCapacity { get; set; }
+
+    private readonly Stream? _stream;
+    private readonly Int32 _bufferSize;
+    private IPacket? _data;
+    private Int32 _total;
+
+    /// <summary>实例化Span读取器。支持从数据流中读取更多数据，突破大小限制</summary>
+    /// <remarks>
+    /// 解析网络协议时，有时候数据帧较大，超过特定缓冲区大小，导致无法一次性读取完整数据帧。
+    /// 加入数据流参数后，在读取数据不足时，SpanReader会自动从数据流中读取一批数据。
+    /// </remarks>
+    /// <param name="stream">数据流。一般是网络流</param>
+    /// <param name="data"></param>
+    /// <param name="bufferSize"></param>
+    public SpanReader(Stream stream, IPacket? data = null, Int32 bufferSize = 8192)
+    {
+        _stream = stream;
+        _bufferSize = bufferSize;
+
+        if (data != null)
+        {
+            _data = data;
+            _span = data.GetSpan();
+            _total = data.Total;
+        }
+    }
     #endregion
 
     #region 基础方法
@@ -69,8 +109,49 @@ public ReadOnlySpan<Byte> GetSpan(Int32 sizeHint = 0)
     /// <summary>确保缓冲区中有足够的空间。</summary>
     /// <param name="size">需要的字节数。</param>
     /// <exception cref="InvalidOperationException"></exception>
-    private void EnsureSpace(Int32 size)
+    public void EnsureSpace(Int32 size)
     {
+        // 检查剩余空间大小，不足时，再从数据流中读取。此时需要注意，创建新的OwnerPacket后，需要先把之前剩余的一点数据拷贝过去，然后再读取Stream
+        var remain = FreeCapacity;
+        if (remain < size && _stream != null)
+        {
+            // 申请指定大小的数据包缓冲区，至少达到缓冲区大小，但不超过最大容量
+            var idx = 0;
+            var bsize = size;
+            if (bsize < _bufferSize) bsize = _bufferSize;
+            if (MaxCapacity > 0 && bsize > MaxCapacity - _total) bsize = MaxCapacity - _total;
+            var pk = new OwnerPacket(bsize);
+            if (_data != null && remain > 0)
+            {
+                if (!_data.TryGetArray(out var arr)) throw new NotSupportedException();
+
+                arr.AsSpan(_index, remain).CopyTo(pk.Buffer);
+                idx += remain;
+            }
+
+            _data.TryDispose();
+            _data = pk;
+            _index = 0;
+
+            // 多次读取，直到满足需求
+            //var n = _stream.ReadExactly(pk.Buffer, pk.Offset + idx, pk.Length - idx);
+            while (idx < size)
+            {
+                // 实际缓冲区大小可能大于申请大小，充分利用缓冲区，避免多次读取
+                var len = pk.Buffer.Length - pk.Offset;
+                var n = _stream.Read(pk.Buffer, pk.Offset + idx, len - idx);
+                if (n <= 0) break;
+
+                idx += n;
+            }
+            if (idx < size)
+                throw new InvalidOperationException("Not enough data to read.");
+            pk.Resize(idx);
+
+            _span = pk.GetSpan();
+            _total += idx - remain;
+        }
+
         if (_index + size > _span.Length)
             throw new InvalidOperationException("Not enough data to read.");
     }
@@ -213,7 +294,6 @@ public String ReadString(Int32 length = 0, Encoding? encoding = null)
     /// <summary>读取字节数组</summary>
     /// <param name="length"></param>
     /// <returns></returns>
-    /// <exception cref="InvalidOperationException"></exception>
     public ReadOnlySpan<Byte> ReadBytes(Int32 length)
     {
         EnsureSpace(length);
@@ -223,6 +303,34 @@ public ReadOnlySpan<Byte> ReadBytes(Int32 length)
         return result;
     }
 
+    /// <summary>读取字节数组</summary>
+    /// <param name="data"></param>
+    /// <returns></returns>
+    public Int32 Read(Span<Byte> data)
+    {
+        var length = data.Length;
+        EnsureSpace(length);
+
+        var result = _span.Slice(_index, length);
+        result.CopyTo(data);
+        _index += length;
+        return length;
+    }
+
+    /// <summary>读取数据包。直接对内部数据包进行切片</summary>
+    /// <param name="length"></param>
+    /// <returns></returns>
+    public IPacket ReadPacket(Int32 length)
+    {
+        if (_data == null) throw new InvalidOperationException("No data stream to read!");
+
+        //EnsureSpace(length);
+
+        var result = _data.Slice(_index, length);
+        _index += length;
+        return result;
+    }
+
     /// <summary>读取结构体</summary>
     /// <typeparam name="T"></typeparam>
     /// <returns></returns>

NewLife.Core/Http/WebSocketMessage.cs

@@ -1,5 +1,4 @@
 using System.Buffers.Binary;
-using System.Drawing;
 using System.Text;
 using NewLife.Buffers;
 using NewLife.Data;
@@ -64,11 +63,7 @@ public Boolean Read(IPacket pk)
     {
         if (pk.Length < 2) return false;
 
-        var reader = new SpanReader(pk.GetSpan())
-        {
-            IsLittleEndian = false
-        };
-        //var reader = pk.GetStream();
+        var reader = new SpanReader(pk) { IsLittleEndian = false };
         var b = reader.ReadByte();
 
         Type = (WebSocketMessageType)(b & 0x7F);
@@ -96,18 +91,18 @@ public Boolean Read(IPacket pk)
         // 如果mask，剩下的就是数据，避免拷贝，提升性能
         if (!mask)
         {
-            Payload = pk.Slice(reader.Position, (Int32)len, true);
+            Payload = reader.ReadPacket((Int32)len);
         }
         else
         {
             var masks = new Byte[4];
-            if (mask) reader.ReadBytes(4).CopyTo(masks);
+            if (mask) reader.Read(masks);
             MaskKey = masks;
 
             if (mask)
             {
                 // 直接在数据缓冲区修改，避免拷贝
-                Payload = pk.Slice(reader.Position, (Int32)len, true);
+                Payload = reader.ReadPacket((Int32)len);
                 var data = Payload.GetSpan();
                 for (var i = 0; i < len; i++)
                 {
@@ -152,10 +147,7 @@ public virtual IPacket ToPacket()
         if (Type == WebSocketMessageType.Close) size += len;
 
         var rs = body.ExpandHeader(size);
-        var writer = new SpanWriter(rs.GetSpan())
-        {
-            IsLittleEndian = false
-        };
+        var writer = new SpanWriter(rs) { IsLittleEndian = false };
 
         writer.WriteByte((Byte)(0x80 | (Byte)Type));
 

NewLife.Core/Messaging/DefaultMessage.cs

@@ -176,14 +176,12 @@ public static Int32 GetLength(ReadOnlySpan<Byte> span)
         reader.Advance(2);
 
         // 小于64k，直接返回
-        //var len = pk.Data.ToUInt16(pk.Offset + 2);
         var len = reader.ReadUInt16();
         if (len < 0xFFFF) return 4 + len;
 
         // 超过64k的超大数据包，再来4个字节
         if (span.Length < 8) return 0;
 
-        //return 8 + (Int32)pk.Data.ToUInt32(pk.Offset + 2 + 2);
         return 8 + reader.ReadInt32();
     }
 

NewLife.Core/Net/Handlers/MessageCodec.cs

@@ -185,54 +185,7 @@ public override Boolean Close(IHandlerContext context, String reason)
     /// <param name="offset">长度的偏移量</param>
     /// <param name="size">长度大小。0变长，1/2/4小端字节，-2/-4大端字节</param>
     /// <returns>数据帧长度（包含头部长度位）</returns>
-    public static Int32 GetLength(IPacket pk, Int32 offset, Int32 size)
-    {
-        if (offset < 0) return pk.Total;
-
-        // 数据不够，连长度都读取不了
-        if (offset >= pk.Total) return 0;
-
-        var reader = new SpanReader(pk.GetSpan());
-        reader.Advance(offset);
-
-        // 读取大小
-        var len = 0;
-        switch (size)
-        {
-            case 0:
-                // 计算变长的头部长度
-                var p = reader.Position;
-                len = reader.ReadEncodedInt() + reader.Position - p;
-                break;
-            case 1:
-                len = reader.ReadByte();
-                break;
-            case 2:
-                len = reader.ReadUInt16();
-                break;
-            case 4:
-                len = reader.ReadInt32();
-                break;
-            case -2:
-                reader.IsLittleEndian = false;
-                len = reader.ReadUInt16();
-                break;
-            case -4:
-                reader.IsLittleEndian = false;
-                len = reader.ReadInt32();
-                break;
-            default:
-                throw new NotSupportedException();
-        }
-
-        // 判断后续数据是否足够
-        if (len > pk.Total) return 0;
-
-        // 数据长度加上头部长度
-        len += Math.Abs(size);
-
-        return offset + len;
-    }
+    public static Int32 GetLength(IPacket pk, Int32 offset, Int32 size) => GetLength(pk.GetSpan(), offset, size);
 
     /// <summary>从数据流中获取整帧数据长度</summary>
     /// <param name="span">数据包</param>
@@ -246,7 +199,7 @@ public static Int32 GetLength(ReadOnlySpan<Byte> span, Int32 offset, Int32 size)
         // 数据不够，连长度都读取不了
         if (offset >= span.Length) return 0;
 
-        var reader = new SpanReader(span);
+        var reader = new SpanReader(span) { IsLittleEndian = true };
         reader.Advance(offset);
 
         // 读取大小

XUnitTest.Core/Buffers/SpanReaderTests.cs

@@ -1,7 +1,6 @@
-using System;
-using System.Text;
-using NewLife;
+using System.Text;
 using NewLife.Buffers;
+using NewLife.Data;
 using NewLife.Security;
 using Xunit;
 
@@ -161,9 +160,62 @@ public void ReadBytesTest()
     {
         var data = new Byte[] { 1, 2, 3, 4, 5 };
         var reader = new SpanReader(data);
+        reader.Advance(1);
 
         var result = reader.ReadBytes(3);
-        Assert.Equal(new Byte[] { 1, 2, 3 }, result.ToArray());
+        Assert.Equal(new Byte[] { 2, 3, 4 }, result.ToArray());
+    }
+
+    [Fact]
+    public void ReadTest()
+    {
+        var data = new Byte[] { 1, 2, 3, 4, 5 };
+        var reader = new SpanReader(data);
+        reader.Advance(1);
+
+        var buf = new Byte[3];
+        var result = reader.Read(buf);
+        Assert.Equal(new Byte[] { 2, 3, 4 }, buf);
+    }
+
+    [Fact]
+    public void ReadPacket()
+    {
+        var data = new Byte[] { 1, 2, 3, 4, 5 };
+        var reader = new SpanReader(data);
+        reader.Advance(1);
+
+        //Assert.Throws<ArgumentOutOfRangeException>(() => reader.ReadPacket(3));
+        try
+        {
+            reader.ReadPacket(3);
+        }
+        catch (Exception ex)
+        {
+
+            Assert.NotNull(ex as InvalidOperationException);
+        }
+
+        var pk = new ArrayPacket(data);
+        reader = new SpanReader(pk);
+        reader.Advance(1);
+
+        var result = reader.ReadPacket(3);
+        Assert.Equal(new Byte[] { 2, 3, 4 }, result.ToArray());
+        Assert.True(data == ((ArrayPacket)result).Buffer);
+    }
+
+    [Fact]
+    public void StreamReadTest()
+    {
+        var data = new Byte[] { 1, 2, 3, 4, 5 };
+        var ms = new MemoryStream(data);
+        var reader = new SpanReader(ms);
+        //reader.Advance(1);
+        reader.ReadByte();
+
+        var result = reader.ReadBytes(3);
+        Assert.Equal(new Byte[] { 2, 3, 4 }, result.ToArray());
     }
 
     [Fact]