Add a new stream only when the connection state is checked

Author: sakno

src/DotNext.Tests/Net/Multiplexing/TcpMultiplexerTests.cs

@@ -267,4 +267,11 @@ public static async Task WaitForDisposedConnectionAsync()
 
         await ThrowsAsync<ObjectDisposedException>(Func.Constant(task));
     }
+    
+    [Fact]
+    public static async Task WaitForCanceledConnectionAsync()
+    {
+        await using var client = new TcpMultiplexedClient(LocalEndPoint, new() { Timeout = DefaultTimeout });
+        await ThrowsAnyAsync<OperationCanceledException>(client.OpenStreamAsync(new CancellationToken(canceled: true)).AsTask);
+    }
 }

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

@@ -89,24 +89,14 @@ public ValueTask StartAsync(CancellationToken token = default)
     /// <seealso cref="DotNext.IO.Pipelines.DuplexStream"/>
     public async ValueTask<IDuplexPipe> OpenStreamAsync(CancellationToken token = default)
     {
-        for (var stream = OpenStream(out var addedStreamId);; token.ThrowIfCancellationRequested())
-        {
-            try
-            {
-                await EnsureConnectedAsync(token).ConfigureAwait(false);
-                return stream;
-            }
-            catch (Exception e)
-            {
-                input.TryRemoveStream(addedStreamId, stream);
-                await stream.AbortAppSideAsync(e).ConfigureAwait(false);
-            }
-        }
+        await EnsureConnectedAsync(token).ConfigureAwait(false);
+        return OpenStream();
     }
 
-    private MultiplexedStream OpenStream(out uint id)
+    private MultiplexedStream OpenStream()
     {
         var stream = new MultiplexedStream(options, writeSignal);
+        uint id;
         do
         {
             id = Interlocked.Increment(ref streamId);