Send Receive Streaming Helpers

src/NexNet/Invocation/IProxyInvoker.cs

@@ -74,4 +74,10 @@
     /// <param name="pipe">Pipe to retrieve the Id of.</param>
     /// <returns>Initial id of the pipe.</returns>
     byte ProxyGetDuplexPipeInitialId(INexusDuplexPipe? pipe);
+
+    IRentedNexusDuplexPipe? GetDuplexPipe();
+
+    ValueTask WriteNexusEnumerableChannel<T>(
+        IRentedNexusDuplexPipe rentedNexusDuplexPipe,
+        NexusEnumerableChannel<T> pipe);
 }

src/NexNet/Invocation/ProxyInvocationBase.cs

@@ -33,7 +33,7 @@ internal CacheManager CacheManager
 
     /// <inheritdoc />
     INexusLogger? IProxyInvoker.Logger => _session?.Logger;
-
+    
     void IProxyInvoker.Configure(
         INexusSession? session,
         SessionManager? sessionManager,
@@ -325,6 +325,19 @@ async ValueTask<TReturn> IProxyInvoker.ProxyInvokeAndWaitForResultCore<TReturn>(
         }
     }
 
+    IRentedNexusDuplexPipe? IProxyInvoker.GetDuplexPipe()
+    {
+        return _session?.PipeManager.RentPipe();
+    }
+
+    ValueTask IProxyInvoker.WriteNexusEnumerableChannel<T>(
+        IRentedNexusDuplexPipe rentedNexusDuplexPipe,
+        NexusEnumerableChannel<T> enumeratorChannel)
+    {
+        enumeratorChannel.DuplexPipe = rentedNexusDuplexPipe;
+        return enumeratorChannel.WriteAndComplete();
+    }
+
     /// <inheritdoc />
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
      byte IProxyInvoker.ProxyGetDuplexPipeInitialId(INexusDuplexPipe? pipe)

src/NexNet/Pipes/INexusDuplexChannel.cs

@@ -1,4 +1,8 @@
 using System;
+using System.Collections.Generic;
+using System.IO.Pipelines;
+using System.Reflection.PortableExecutable;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace NexNet.Pipes;
@@ -40,3 +44,117 @@
     /// <returns>A ValueTask that represents the asynchronous operation. The task result contains the <see cref="INexusChannelReader{T}"/> instance.</returns>
     ValueTask<INexusChannelReader<T>> GetReaderAsync();
 }
+
+/// <summary>
+/// Represents a stream of objects that can be asynchronously enumerated.
+/// </summary>
+/// <typeparam name="T">The type of objects in the stream.</typeparam>
+public class NexusEnumerableChannel<T> :IAsyncEnumerable<T>
+{
+    private readonly IEnumerable<T>? _writingEnumerable;
+    internal IRentedNexusDuplexPipe DuplexPipe;
+
+    /// <summary>
+    /// Initializes a new instance of the NexusEnumerableStream class.
+    /// </summary>
+    public NexusEnumerableChannel(IEnumerable<T> writingEnumerable)
+    {
+        _writingEnumerable = writingEnumerable;
+    }
+
+    internal NexusEnumerableChannel(IRentedNexusDuplexPipe duplexPipe)
+    {
+        _writingEnumerable = null;
+        DuplexPipe = duplexPipe;
+    }
+
+    internal async ValueTask WriteAndComplete()
+    {
+        var writer = await DuplexPipe.GetChannelWriter<T>();
+        await writer.WriteAndComplete(_writingEnumerable);
+    }
+
+    public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = new CancellationToken())
+    {
+        return new Enumerator(DuplexPipe);
+    }
+
+    private class Enumerator : IAsyncEnumerator<T>
+    {
+        private NexusChannelReader<T>? _reader;
+        private readonly INexusDuplexPipe _duplexPipe;
+        private List<T>? _list = null;
+        private int _readIndex = 0;
+
+        /// <summary>
+        /// Creates an instance of the Enumerator class with the specified IRentedNexusDuplexPipe object.
+        /// </summary>
+        /// <param name="duplexPipe">The IRentedNexusDuplexPipe object to use for reading data.</param>
+        public Enumerator(IRentedNexusDuplexPipe duplexPipe)
+        {
+            _duplexPipe = duplexPipe;
+            _reader = new NexusChannelReader<T>(_duplexPipe.ReaderCore);
+        }
+
+        /// <summary>
+        /// Gets or sets the current value of the property.
+        /// </summary>
+        /// <value>The current value of the property.</value>
+        public T Current { get; set; }
+
+        public async ValueTask<bool> MoveNextAsync()
+        {
+            // If the reader is null, the channel has been read to completion.
+            if (_reader == null)
+            {
+                // If the read index is -1, then the pipe reading has completed.
+                if (_readIndex == -1)
+                    return false;
+
+                if (_readIndex < _list!.Count)
+                {
+                    Current = _list[_readIndex++];
+                    return true;
+                }
+
+                // If we got here, we are at the end of the list and the channel has completed.
+                Current = default!;
+                _readIndex = -1;
+                _list.Clear();
+                _list.TrimExcess();
+                _list = null;
+
+                return false;
+            }
+
+            if (_list == null)
+            {
+                await _duplexPipe.ReadyTask;
+                _list = new List<T>();
+            }
+
+            if (_readIndex < _list.Count)
+            {
+                Current = _list[_readIndex++];
+                return true;
+            }
+
+            var result = await _reader.ReadAsync(_list, null);
+
+            // If false, then the reader has completed.
+            if (result == false)
+            {
+                _readIndex = -1;
+                return false;
+            }
+
+            _readIndex = 1;
+            Current = _list[_readIndex];
+            return true;
+        }
+        public ValueTask DisposeAsync()
+        {
+            return _duplexPipe.Input.CompleteAsync();
+        }
+    }
+}

src/NexNet/Pipes/RentedNexusDuplexPipe.cs

@@ -15,11 +15,6 @@ public RentedNexusDuplexPipe(byte localId, INexusSession session)
 
     public ValueTask DisposeAsync()
     {
-        var manager = Interlocked.Exchange(ref Manager, null);
-
-        if (manager == null)
-            return default;
-
-        return manager!.ReturnPipe(this);
+        return Interlocked.Exchange(ref Manager, null)?.ReturnPipe(this) ?? default;
     }
 }