Add ToReadOnlySequence() to ByteString

src/core/Akka.Tests/Util/ByteStringSpec.cs

@@ -6,6 +6,7 @@
 //-----------------------------------------------------------------------
 
 using System;
+using System.Buffers;
 using System.Linq;
 using System.Runtime.InteropServices;
 using System.Text;
@@ -97,6 +98,62 @@ public void A_ByteString_ToReadOnlySpan_compacted_must_have_identical_memory_ref
             var concatSpan4 = compacted.ToReadOnlySpan();
             (concatSpan3 == concatSpan4).Should().BeTrue();
         }
+
+        [Fact]
+        public void A_ByteString_ToReadOnlySequence_must_have_correct_size()
+        {
+            Prop.ForAll((ByteString a, ByteString b) =>
+            {
+                a.ToReadOnlySequence().Length.Should().Be(a.Count);
+                b.ToReadOnlySequence().Length.Should().Be(b.Count);
+                var concat = a + b;
+                var spanConcat = concat.ToReadOnlySequence();
+                return spanConcat.Length == concat.Count;
+            }).QuickCheckThrowOnFailure();
+        }
+
+        [Fact]
+        public void A_ByteString_ToReadOnlySequence_compacted_must_have_identical_contents_to_original()
+        {
+            var bytesA = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+            var bytesB = new byte[] { 10, 11, 12, 13, 14, 15, 16, 17, 18 };
+            var a = ByteString.FromBytes(bytesA);
+            var b = ByteString.FromBytes(bytesB);
+
+            var concat = a + b;
+
+            var concatSpan = concat.ToReadOnlySequence();
+            var concatSpan2 = concat.ToReadOnlySequence();
+            // not compact, returns a new span
+            (concatSpan.ToArray().SequenceEqual(concatSpan2.ToArray())).Should().BeTrue();
+
+            // compact, will return same span
+            var compacted = concat.Compact();
+            var concatSpan3 = compacted.ToReadOnlySequence();
+            (concatSpan.ToArray().SequenceEqual(concatSpan3.ToArray())).Should().BeTrue();
+        }
+
+        [Fact]
+        public void A_ByteString_ToReadOnlySequence_compacted_must_have_identical_contents_to_original_with_many_Segments()
+        {
+            var bytesA = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+            var bytesB = new byte[] { 10, 11, 12, 13, 14, 15, 16, 17, 18 };
+            var bytesC = new byte[] { 19, 20, 21, 22, 23, 24, 25, 26, 27 };
+            var a = ByteString.FromBytes(bytesA);
+            var b = ByteString.FromBytes(bytesB);
+            var c = ByteString.FromBytes(bytesC);
+            var concat = a + b +c;
+
+            var concatSpan = concat.ToReadOnlySequence();
+            var concatSpan2 = concat.ToReadOnlySequence();
+            // not compact, returns a new span
+            (concatSpan.ToArray().SequenceEqual(concatSpan2.ToArray())).Should().BeTrue();
+
+            // compact, will return same span
+            var compacted = concat.Compact();
+            var concatSpan3 = compacted.ToReadOnlySequence();
+            (concatSpan.ToArray().SequenceEqual(concatSpan3.ToArray())).Should().BeTrue();
+        }
 
         [Fact]
         public void A_ByteString_must_have_correct_size_when_slicing_from_index()

src/core/Akka/Util/ByteString.cs

@@ -6,6 +6,7 @@
 //-----------------------------------------------------------------------
 
 using System;
+using System.Buffers;
 using System.Collections;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -28,6 +29,46 @@ namespace Akka.IO
     [DebuggerDisplay("(Count = {_count}, Buffers = {_buffers})")]
     public sealed class ByteString : IEquatable<ByteString>, IEnumerable<byte>
     {
+        public class ByteStringReadOnlySequenceSegment : ReadOnlySequenceSegment<byte>
+        {
+            public ByteStringReadOnlySequenceSegment(ReadOnlyMemory<byte> memory, long runningIndex)
+            {
+                Memory = memory;
+                RunningIndex = runningIndex;
+            }
+
+            /// <remarks>
+            /// This is here because <see cref="ReadOnlySequenceSegment{T}"/>
+            /// has predefined properties with Protected Setters.
+            /// </remarks>
+            public static ReadOnlySequence<byte> CreateSequence(ByteString bs)
+            {
+                var bArr = bs._buffers;
+                var first = new ByteStringReadOnlySequenceSegment(bArr[0],0);
+                ByteBuffer item = default;
+                ByteStringReadOnlySequenceSegment last = first;
+                if (bArr.Length == 2)
+                {
+                    item = bArr[1];
+                    last = new ByteStringReadOnlySequenceSegment(item, bs.Count-item.Count);
+                    first.Next = last;
+                }
+                else
+                {
+                    var prior = first;
+                    for (int i = 1; i < bArr.Length; i++)
+                    {
+                        item = bArr[i];
+                        var curr = new ByteStringReadOnlySequenceSegment(item, prior.RunningIndex+prior.Memory.Length);
+                        prior.Next = curr;
+                        prior = curr;
+                    }
+                    last = prior;
+                }
+                return new ReadOnlySequence<byte>(first,0,last,last.Memory.Length);
+            }
+        }
+
         #region creation methods
 
         /// <summary>
@@ -246,7 +287,7 @@ public static ByteString FromString(string str, Encoding encoding)
 
         private readonly int _count;
         private readonly ByteBuffer[] _buffers;
-
+        
         private ByteString(ByteBuffer[] buffers, int count)
         {
             _buffers = buffers;
@@ -528,6 +569,30 @@ public ReadOnlySpan<byte> ToReadOnlySpan()
 
             return new ReadOnlySpan<byte>(ToArray());
         }
+
+        /// <summary>
+        /// Returns a ReadOnlySequence<byte/> over the contents of this ByteString.
+        /// This is not a copying operation and zero-alloc when ByteString is compact 
+        /// Otherwise N <see cref="ByteStringReadOnlySequenceSegment"/>s will be allocated
+        /// Where N is the number of arrays currently internally held by the ByteString
+        /// </summary>
+        /// <returns>A ReadOnlySpan<byte/> over the byte data.</returns>
+        public ReadOnlySequence<byte> ToReadOnlySequence()
+        {
+            if (_count == 0)
+            {
+                return ReadOnlySequence<byte>.Empty;
+            }
+            else if (_buffers.Length == 1)
+            {
+                //Happy path, we can just pass ArraySegment here and avoid alloc.
+                return new ReadOnlySequence<byte>(_buffers[0]);
+            }
+            else
+            {
+                return ByteStringReadOnlySequenceSegment.CreateSequence(this);
+            }
+        }
 
         /// <summary>
         /// Appends <paramref name="other"/> <see cref="ByteString"/> at the tail