Fixed ObjectDisposedException

Author: sakno

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/InputMultiplexer.cs

@@ -42,7 +42,7 @@ public async Task ProcessAsync(Func<bool> condition, Socket socket)
     {
         using var enumerator = streams.GetEnumerator();
         for (var requiresHeartbeat = false;
-             !Token.IsCancellationRequested && condition();
+             condition();
              requiresHeartbeat = !await writeSignal.WaitAsync(heartbeatTimeout, Token).ConfigureAwait(false))
         {
             framingBuffer.Clear(reuseBuffer: true);
@@ -97,7 +97,7 @@ private async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Socket socket)
             timeoutSource.Start(timeout);
             try
             {
-                bytesWritten = await socket.SendAsync(buffer, SocketFlags.None).ConfigureAwait(false);
+                bytesWritten = await socket.SendAsync(buffer, SocketFlags.None, timeoutSource.Token).ConfigureAwait(false);
             }
             catch (OperationCanceledException e) when (timeoutSource.IsTimedOut(e))
             {

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedClient.Dispatcher.cs

@@ -31,7 +31,7 @@ private async Task DispatchAsync()
         var receiveLoop = Task.CompletedTask;
 
         // send loop
-        while (!input.Token.IsCancellationRequested)
+        while (true)
         {
             try
             {

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedListener.Dispatcher.cs

@@ -52,7 +52,7 @@ private async Task DispatchAsync(Socket socket)
         {
             var receiveLoop = output.ProcessAsync(socket);
 
-            while (!lifetimeToken.IsCancellationRequested)
+            while (true)
             {
                 // rethrow exception from the receiving loop
                 if (receiveLoop.IsCompleted)

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/OutputMultiplexer.cs

@@ -33,15 +33,15 @@ public Task ProcessAsync(Socket socket)
     private async Task ProcessCoreAsync(Socket socket)
     {
         FrameHeader header;
-        for (var bufferedBytes = 0; !Token.IsCancellationRequested; AdjustFramingBuffer(ref bufferedBytes, header, framingBuffer.Span))
+        for (var bufferedBytes = 0;; AdjustFramingBuffer(ref bufferedBytes, header, framingBuffer.Span))
         {
             timeoutSource.Start(timeout); // resumed by heartbeat
             try
             {
                 // read at least header
                 while (bufferedBytes < FrameHeader.Size)
                 {
-                    bufferedBytes += await socket.ReceiveAsync(framingBuffer.Slice(bufferedBytes), Token).ConfigureAwait(false);
+                    bufferedBytes += await socket.ReceiveAsync(framingBuffer.Slice(bufferedBytes), timeoutSource.Token).ConfigureAwait(false);
                 }
 
                 header = FrameHeader.Parse(framingBuffer.Span);
@@ -91,7 +91,9 @@ private async Task ProcessCoreAsync(Socket socket)
             }
 
             // write the frame to the output header
-            await stream.ReadFrameAsync(header.Control, framingBuffer.Slice(FrameHeader.Size, header.Length), Token).ConfigureAwait(false);
+            await stream
+                .ReadFrameAsync(header.Control, framingBuffer.Slice(FrameHeader.Size, header.Length), Token)
+                .ConfigureAwait(false);
         }
     }