Add asynchronous Stream serializer APIs

src/LightProto/PooledSegmentBufferWriter.cs

@@ -0,0 +1,193 @@
+using System.Buffers;
+using System.Collections.Concurrent;
+
+namespace LightProto
+{
+    /// <summary>
+    /// A pooled, segmented buffer used to bridge synchronous codecs with asynchronous stream I/O.
+    /// </summary>
+    internal sealed class PooledSegmentBufferWriter : IBufferWriter<byte>
+    {
+        private const int MinimumSegmentSize = 4096;
+
+        private static readonly ConcurrentBag<PooledSegmentBufferWriter> WriterPool = new();
+        private static readonly ConcurrentBag<Segment> SegmentPool = new();
+
+        private Segment? first;
+        private Segment? current;
+
+        private PooledSegmentBufferWriter() { }
+
+        public static PooledSegmentBufferWriter Rent()
+        {
+            if (WriterPool.TryTake(out var writer))
+            {
+                return writer;
+            }
+
+            return new PooledSegmentBufferWriter();
+        }
+
+        public static void Return(PooledSegmentBufferWriter writer)
+        {
+            writer.Reset();
+            WriterPool.Add(writer);
+        }
+
+        public void Advance(int count)
+        {
+            if (current is null)
+            {
+                if (count == 0)
+                {
+                    return;
+                }
+
+                throw new InvalidOperationException("No buffer has been requested.");
+            }
+
+            var segment = current;
+            if ((uint)count > (uint)(segment.Buffer.Length - segment.Written))
+            {
+                throw new ArgumentOutOfRangeException(nameof(count));
+            }
+
+            segment.Written += count;
+        }
+
+        public Memory<byte> GetMemory(int sizeHint = 0)
+        {
+            var segment = GetWritableSegment(sizeHint);
+            return segment.Buffer.AsMemory(segment.Written);
+        }
+
+        public Span<byte> GetSpan(int sizeHint = 0)
+        {
+            var segment = GetWritableSegment(sizeHint);
+            return segment.Buffer.AsSpan(segment.Written);
+        }
+
+        public async Task ReadToEndAsync(Stream source, CancellationToken cancellationToken)
+        {
+            while (true)
+            {
+                var segment = GetWritableSegment(MinimumSegmentSize);
+                var read = await source
+                    .ReadAsync(segment.Buffer, segment.Written, segment.Buffer.Length - segment.Written, cancellationToken)
+                    .ConfigureAwait(false);
+                if (read == 0)
+                {
+                    return;
+                }
+
+                segment.Written += read;
+            }
+        }
+
+        public async Task WriteToAsync(Stream destination, CancellationToken cancellationToken)
+        {
+            for (var segment = first; segment is not null; segment = segment.NextSegment)
+            {
+                if (segment.Written == 0)
+                {
+                    continue;
+                }
+
+                await destination.WriteAsync(segment.Buffer, 0, segment.Written, cancellationToken).ConfigureAwait(false);
+            }
+        }
+
+        public ReadOnlySequence<byte> GetReadOnlySequence()
+        {
+            if (first is null || current is null)
+            {
+                return ReadOnlySequence<byte>.Empty;
+            }
+
+            for (var segment = first; segment is not null; segment = segment.NextSegment)
+            {
+                segment.SetMemory();
+            }
+
+            return new ReadOnlySequence<byte>(first, 0, current, current.Written);
+        }
+
+        private Segment GetWritableSegment(int sizeHint)
+        {
+            if (sizeHint < 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(sizeHint));
+            }
+
+            if (current is null || current.Buffer.Length - current.Written < Math.Max(sizeHint, 1))
+            {
+                var segment = RentSegment(Math.Max(sizeHint, MinimumSegmentSize));
+                if (current is null)
+                {
+                    first = segment;
+                }
+                else
+                {
+                    segment.SetRunningIndex(current.RunningIndex + current.Written);
+                    current.NextSegment = segment;
+                }
+
+                current = segment;
+            }
+
+            return current;
+        }
+
+        private void Reset()
+        {
+            var segment = first;
+            first = null;
+            current = null;
+
+            while (segment is not null)
+            {
+                var next = segment.NextSegment;
+                segment.Reset();
+                SegmentPool.Add(segment);
+                segment = next;
+            }
+        }
+
+        private static Segment RentSegment(int minimumSize)
+        {
+            if (!SegmentPool.TryTake(out var segment))
+            {
+                segment = new Segment();
+            }
+
+            segment.Buffer = ArrayPool<byte>.Shared.Rent(minimumSize);
+            return segment;
+        }
+
+        private sealed class Segment : ReadOnlySequenceSegment<byte>
+        {
+            public byte[] Buffer = null!;
+            public int Written;
+
+            public Segment? NextSegment
+            {
+                get => (Segment?)Next;
+                set => Next = value;
+            }
+
+            public void SetMemory() => Memory = Buffer.AsMemory(0, Written);
+
+            public void SetRunningIndex(long runningIndex) => RunningIndex = runningIndex;
+
+            public void Reset()
+            {
+                ArrayPool<byte>.Shared.Return(Buffer);
+                Buffer = null!;
+                Written = 0;
+                Memory = default;
+                Next = null;
+                RunningIndex = 0;
+            }
+        }
+    }
+}

src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt

@@ -1,4 +1,15 @@
 #nullable enable
+static LightProto.Serializer.DeserializeAsync<T>(System.IO.Stream! source, LightProto.IProtoReader<T>! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<T>!
+static LightProto.Serializer.DeserializeAsync<T>(System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<T>!
+static LightProto.Serializer.DeserializeDynamicallyAsync<T>(System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<T>!
+static LightProto.Serializer.DeserializeNonGenericAsync(System.IO.Stream! source, LightProto.IProtoReader! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<object!>!
+static LightProto.Serializer.DeserializeNonGenericAsync(System.Type! type, System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<object!>!
+static LightProto.Serializer.SerializeAsync<T>(System.IO.Stream! destination, T instance, LightProto.IProtoWriter<T>! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeAsync<T>(System.IO.Stream! destination, T instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeDynamicallyAsync<T>(System.IO.Stream! destination, T instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeToAsync<T>(this T instance, System.IO.Stream! destination, LightProto.IProtoWriter<T>! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
 LightProto.IProtoWriter.CalculateLongSize(object! value) -> long
 LightProto.IProtoWriter<T>.CalculateLongSize(T value) -> long
 LightProto.PackedRepeatedOptimizer

src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt

@@ -1,4 +1,13 @@
 #nullable enable
+static LightProto.Serializer.DeserializeAsync<T>(System.IO.Stream! source, LightProto.IProtoReader<T>! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<T>!
+static LightProto.Serializer.DeserializeDynamicallyAsync<T>(System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<T>!
+static LightProto.Serializer.DeserializeNonGenericAsync(System.IO.Stream! source, LightProto.IProtoReader! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<object!>!
+static LightProto.Serializer.DeserializeNonGenericAsync(System.Type! type, System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<object!>!
+static LightProto.Serializer.SerializeAsync<T>(System.IO.Stream! destination, T instance, LightProto.IProtoWriter<T>! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeDynamicallyAsync<T>(System.IO.Stream! destination, T instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
+static LightProto.Serializer.SerializeToAsync<T>(this T instance, System.IO.Stream! destination, LightProto.IProtoWriter<T>! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
 LightProto.IProtoWriter.CalculateLongSize(object! value) -> long
 LightProto.IProtoWriter<T>.CalculateLongSize(T value) -> long
 LightProto.PackedRepeatedOptimizer

src/LightProto/Serializer.Async.cs

@@ -0,0 +1,90 @@
+using System.Runtime.CompilerServices;
+
+namespace LightProto
+{
+#pragma warning disable RS0026 // CancellationToken is intentionally optional for the asynchronous API surface.
+    public static partial class Serializer
+    {
+        /// <summary>
+        /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream.
+        /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins.
+        /// </summary>
+        public static async Task SerializeAsync<T>(
+            Stream destination,
+            T instance,
+            IProtoWriter<T> writer,
+            CancellationToken cancellationToken = default
+        )
+        {
+            if (destination is null)
+            {
+                throw new ArgumentNullException(nameof(destination));
+            }
+            if (writer is null)
+            {
+                throw new ArgumentNullException(nameof(writer));
+            }
+
+            var buffer = PooledSegmentBufferWriter.Rent();
+            try
+            {
+                Serialize(buffer, instance, writer);
+                await buffer.WriteToAsync(destination, cancellationToken).ConfigureAwait(false);
+            }
+            finally
+            {
+                PooledSegmentBufferWriter.Return(buffer);
+            }
+        }
+
+        /// <summary>
+        /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream.
+        /// The accumulated message is decoded synchronously after all input has been received.
+        /// </summary>
+        public static async Task<T> DeserializeAsync<T>(
+            Stream source,
+            IProtoReader<T> reader,
+            CancellationToken cancellationToken = default
+        )
+        {
+            if (source is null)
+            {
+                throw new ArgumentNullException(nameof(source));
+            }
+            if (reader is null)
+            {
+                throw new ArgumentNullException(nameof(reader));
+            }
+
+            var buffer = PooledSegmentBufferWriter.Rent();
+            try
+            {
+                await buffer.ReadToEndAsync(source, cancellationToken).ConfigureAwait(false);
+                return Deserialize(buffer.GetReadOnlySequence(), reader);
+            }
+            finally
+            {
+                PooledSegmentBufferWriter.Return(buffer);
+            }
+        }
+
+#if NET7_0_OR_GREATER
+        /// <summary>
+        /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream.
+        /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins.
+        /// </summary>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public static Task SerializeAsync<T>(Stream destination, T instance, CancellationToken cancellationToken = default)
+            where T : IProtoParser<T> => SerializeAsync(destination, instance, T.ProtoWriter, cancellationToken);
+
+        /// <summary>
+        /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream.
+        /// The accumulated message is decoded synchronously after all input has been received.
+        /// </summary>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public static Task<T> DeserializeAsync<T>(Stream source, CancellationToken cancellationToken = default)
+            where T : IProtoParser<T> => DeserializeAsync(source, T.ProtoReader, cancellationToken);
+#endif
+    }
+#pragma warning restore RS0026
+}

src/LightProto/Serializer.Dynamically.cs

@@ -43,6 +43,20 @@ public static void SerializeDynamically<
 #endif
             T>(Stream destination, T instance) => Serialize(destination, instance, GetProtoWriter<T>());
 
+        /// <summary>
+        /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream.
+        /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins.
+        /// </summary>
+#if NET7_0_OR_GREATER
+        [RequiresDynamicCode(AOTWarning)]
+#endif
+        public static Task SerializeDynamicallyAsync<
+#if NET7_0_OR_GREATER
+            [DynamicallyAccessedMembers(LightProtoRequiredMembers)]
+#endif
+            T>(Stream destination, T instance, CancellationToken cancellationToken = default) =>
+            SerializeAsync(destination, instance, GetProtoWriter<T>(), cancellationToken);
+
         /// <summary>
         /// Creates a new instance from a protocol-buffer stream
         /// </summary>
@@ -58,6 +72,20 @@ public static T DeserializeDynamically<
 #endif
             T>(Stream source) => Deserialize(source, GetProtoReader<T>());
 
+        /// <summary>
+        /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream.
+        /// The accumulated message is decoded synchronously after all input has been received.
+        /// </summary>
+#if NET7_0_OR_GREATER
+        [RequiresDynamicCode(AOTWarning)]
+#endif
+        public static Task<T> DeserializeDynamicallyAsync<
+#if NET7_0_OR_GREATER
+            [DynamicallyAccessedMembers(LightProtoRequiredMembers)]
+#endif
+            T>(Stream source, CancellationToken cancellationToken = default) =>
+            DeserializeAsync(source, GetProtoReader<T>(), cancellationToken);
+
         /// <summary>
         /// Serializes the given message to a byte array.
         /// </summary>

src/LightProto/Serializer.Extensions.cs

@@ -178,6 +178,17 @@ public static byte[] ToByteArray<T>(this T message, IProtoWriter<T> writer)
         public static void SerializeTo<T>(this T instance, Stream destination, IProtoWriter<T> writer) =>
             Serialize(destination, instance, writer);
 
+        /// <summary>
+        /// Asynchronously serializes the instance to the given destination stream.
+        /// The instance is encoded synchronously into a pooled buffer before asynchronous I/O begins.
+        /// </summary>
+        public static Task SerializeToAsync<T>(
+            this T instance,
+            Stream destination,
+            IProtoWriter<T> writer,
+            CancellationToken cancellationToken = default
+        ) => SerializeAsync(destination, instance, writer, cancellationToken);
+
         /// <summary>
         /// Serializes the instance to the given destination buffer.
         /// </summary>