fixed framing issues

Author: Aaronontheweb

src/core/Akka.Streams/Dsl/Framing.cs

@@ -169,39 +169,45 @@ public FramingException(string message) : base(message)
             protected FramingException(SerializationInfo info, StreamingContext context) : base(info, context) { }
         }
 
-        private static readonly Func<IEnumerator<byte>, int, int> BigEndianDecoder = (enumerator, length) =>
+        private static int BigEndianDecoder(ref ReadOnlySpan<byte> enumerator, int length)
         {
-            var count = length;
-            var decoded = 0;
-            while (count > 0)
+            if (length > enumerator.Length)
+            {
+                throw new ArgumentException("Length exceeds the size of the enumerator.");
+            }
+
+            var result = 0;
+
+            // Assuming 'length' is 4 for a 32-bit integer.
+            // Adjust the loop and shift amounts if dealing with different sizes.
+            for (var i = 0; i < length; i++)
             {
-                decoded <<= 8;
-                if (!enumerator.MoveNext()) throw new IndexOutOfRangeException("LittleEndianDecoder reached end of byte string");
-                decoded |= enumerator.Current & 0xFF;
-                count--;
+                result |= enumerator[i] << ((length - 1 - i) * 8);
             }
 
-            return decoded;
-        };
+            return result;
+        }
 
-        private static readonly Func<IEnumerator<byte>, int, int> LittleEndianDecoder = (enumerator, length) =>
+        private static int LittleEndianDecoder(ref ReadOnlySpan<byte> span, int length)
         {
-            var highestOcted = (length - 1) << 3;
-            var mask = (int) (1L << (length << 3)) - 1;
-            var count = length;
+            if (length > span.Length)
+            {
+                throw new IndexOutOfRangeException("LittleEndianDecoder reached end of byte span");
+            }
+
             var decoded = 0;
+            var highestOctetShift = (length - 1) << 3;
+            var mask = (int)(1L << (length << 3)) - 1;
 
-            while (count > 0)
+            for (var i = 0; i < length; i++)
             {
-                // decoded >>>= 8 on the jvm
-                decoded = (int) ((uint) decoded >> 8);
-                if (!enumerator.MoveNext()) throw new IndexOutOfRangeException("LittleEndianDecoder reached end of byte string");
-                decoded += (enumerator.Current & 0xFF) << highestOcted;
-                count--;
+                // Shift and add the ith byte to 'decoded'. No need for >>>= as in JVM; just shift appropriately.
+                var shiftAmount = highestOctetShift - (i << 3);
+                decoded |= (span[i] & 0xFF) << shiftAmount;
             }
 
             return decoded & mask;
-        };
+        }
 
         private sealed class SimpleFramingProtocolEncoderStage : SimpleLinearGraphStage<ByteString>
         {
@@ -342,8 +348,8 @@ private void DoParse()
                         else if (_buffer.HasSubstring(_stage._separatorBytes, possibleMatchPosition))
                         {
                             // Found a match
-                            var parsedFrame = _buffer.Slice(0, possibleMatchPosition).Compact();
-                            _buffer = _buffer.Slice(possibleMatchPosition + _stage._separatorBytes.Count).Compact();
+                            var parsedFrame = _buffer.Slice(0, possibleMatchPosition);
+                            _buffer = _buffer.Slice(possibleMatchPosition + _stage._separatorBytes.Count);
                             _nextPossibleMatch = 0;
                             Push(_stage.Outlet, parsedFrame);
 
@@ -422,7 +428,7 @@ public override void OnUpstreamFinish()
                 /// </summary>
                 private void PushFrame()
                 {
-                    var emit = _buffer.Slice(0, _frameSize).Compact();
+                    var emit = _buffer.Slice(0, _frameSize);
                     _buffer = _buffer.Slice(_frameSize);
                     _frameSize = int.MaxValue;
                     Push(_stage.Outlet, emit);
@@ -440,9 +446,14 @@ private void TryPushFrame()
                         PushFrame();
                     else if (bufferSize >= _stage._minimumChunkSize)
                     {
-                        var iterator = _buffer.Slice(_stage._lengthFieldOffset).GetEnumerator();
-                        var parsedLength = _stage._intDecoder(iterator, _stage._lengthFieldLength);
-
+                        var iterator = _buffer.Memory.Span.Slice(_stage._lengthFieldOffset);
+                        var parsedLength = _stage._byteOrder switch {
+                            ByteOrder.BigEndian => BigEndianDecoder(ref iterator, _stage._lengthFieldLength),
+                            ByteOrder.LittleEndian => LittleEndianDecoder(ref iterator, _stage._lengthFieldLength),
+                            _ => throw new NotSupportedException($"ByteOrder {_stage._byteOrder} is not supported")
+                        }; 
+
+                        // TODO: AVOID ARRAY COPYING AGAIN HERE
                         _frameSize = _stage._computeFrameSize.HasValue
                             ? _stage._computeFrameSize.Value(_buffer.Slice(0, _stage._lengthFieldOffset).ToArray(), parsedLength)
                             : parsedLength + _stage._minimumChunkSize;
@@ -480,7 +491,7 @@ private void TryPull()
             private readonly int _maximumFramelength;
             private readonly int _lengthFieldOffset;
             private readonly int _minimumChunkSize;
-            private readonly Func<IEnumerator<byte>, int, int> _intDecoder;
+            private readonly ByteOrder _byteOrder;
             private readonly Option<Func<IReadOnlyList<byte>, int, int>> _computeFrameSize;
 
             // For the sake of binary compatibility
@@ -500,7 +511,7 @@ public LengthFieldFramingStage(
                 _lengthFieldOffset = lengthFieldOffset;
                 _minimumChunkSize = lengthFieldOffset + lengthFieldLength;
                 _computeFrameSize = computeFrameSize;
-                _intDecoder = byteOrder == ByteOrder.BigEndian ? BigEndianDecoder : LittleEndianDecoder;
+                _byteOrder = byteOrder;
             }
 
             protected override GraphStageLogic CreateLogic(Attributes inheritedAttributes) => new Logic(this);