Improving the connection logic.

Author: Stefán J. Sigurðarson

RabbitMQTest/Program.cs

@@ -15,6 +15,7 @@ class Program
         private static int itemsPerBatch = 500;
         static async Task Main(string[] args)
         {
+            ThreadPool.SetMinThreads(16 * Environment.ProcessorCount, 16 * Environment.ProcessorCount);
             var connectionString = new Uri("amqp://guest:guest@localhost/");
 
             var connectionFactory = new ConnectionFactory() { DispatchConsumersAsync = true, Uri = connectionString };
@@ -23,7 +24,7 @@ static async Task Main(string[] args)
             var publisher = connection.CreateModel();
             var subscriber = connection2.CreateModel();
             publisher.ConfirmSelect();
-            subscriber.ConfirmSelect();
+            //subscriber.ConfirmSelect();
 
             publisher.ExchangeDeclare("test", ExchangeType.Topic, true);
 
@@ -47,8 +48,8 @@ static async Task Main(string[] args)
                         batch.Add("test", "myawesome.routing.key", false, properties, payload);
                     }
                     batch.Publish();
-                    await publisher.WaitForConfirmsOrDieAsync();
                     messagesSent += itemsPerBatch;
+                    await publisher.WaitForConfirmsOrDieAsync();
                 }
             });
 

projects/client/RabbitMQ.Client/src/client/api/SslHelper.cs

@@ -72,9 +72,7 @@ public static Stream TcpUpgrade(Stream tcpStream, SslOption sslOption)
 
             var sslStream = new SslStream(tcpStream, false, remoteCertValidator, localCertSelector);
 
-            sslStream.AuthenticateAsClientAsync(sslOption.ServerName, sslOption.Certs, sslOption.Version,
-                                                sslOption.CheckCertificateRevocation).GetAwaiter().GetResult();
-
+            sslStream.AuthenticateAsClient(sslOption.ServerName, sslOption.Certs, sslOption.Version, sslOption.CheckCertificateRevocation);
             return sslStream;
         }
 

projects/client/RabbitMQ.Client/src/client/impl/AsyncRpcContinuation.cs

@@ -0,0 +1,53 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using RabbitMQ.Client.Exceptions;
+
+namespace RabbitMQ.Client.Impl
+{
+    public class AsyncRpcContinuation : IRpcContinuation
+    {
+        private readonly TaskCompletionSource<Command> _taskCompletionSource = new TaskCompletionSource<Command>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public virtual async ValueTask<Command> GetReplyAsync(CancellationToken cancellationToken = default)
+        {
+            try
+            {
+                if (cancellationToken != default)
+                {
+                    using (cancellationToken.Register(() => _taskCompletionSource.TrySetCanceled(cancellationToken)))
+                    {
+                        return await _taskCompletionSource.Task.ConfigureAwait(false);
+                    }
+                }
+
+                return await _taskCompletionSource.Task.ConfigureAwait(false);
+            }
+            catch (OperationInterruptedException)
+            {
+                throw;
+            }
+        }
+
+        public virtual async ValueTask<Command> GetReplyAsync(TimeSpan timeout)
+        {
+            using (CancellationTokenSource cts = new CancellationTokenSource(timeout))
+            {
+                return await GetReplyAsync(cts.Token).ConfigureAwait(false);
+            }
+        }
+
+        public void HandleCommand(Command cmd)
+        {
+            _taskCompletionSource.TrySetResult(cmd);
+        }
+
+        public void HandleModelShutdown(ShutdownEventArgs reason)
+        {
+            _taskCompletionSource.TrySetException(new OperationInterruptedException(reason));
+        }
+    }
+}

projects/client/RabbitMQ.Client/src/client/impl/Connection.cs

@@ -1181,19 +1181,25 @@ Command ChannelCloseWrapper(ushort reasonCode, string reasonText)
 
         void StartAndTune()
         {
-            var connectionStartCell = new BlockingCell<ConnectionStartDetails>();
-            m_model0.m_connectionStartCell = connectionStartCell;
+            m_model0.m_connectionStartCell = new TaskCompletionSource<ConnectionStartDetails>(TaskCreationOptions.RunContinuationsAsynchronously);
             m_model0.HandshakeContinuationTimeout = m_factory.HandshakeContinuationTimeout;
             m_frameHandler.ReadTimeout = m_factory.HandshakeContinuationTimeout;
             m_frameHandler.SendHeader();
+            try
+            {
+                m_model0.m_connectionStartCell.Task.Wait(m_factory.HandshakeContinuationTimeout);
+            }
+            catch (Exception)
+            {
+                m_model0.m_connectionStartCell.TrySetCanceled();
+            }
 
-            var connectionStart = connectionStartCell.WaitForValue();
-
-            if (connectionStart == null)
+            if (m_model0.m_connectionStartCell.Task.IsCanceled)
             {
                 throw new IOException("connection.start was never received, likely due to a network timeout");
             }
 
+            var connectionStart = m_model0.m_connectionStartCell.Task.Result;
             ServerProperties = connectionStart.m_serverProperties;
 
             var serverVersion = new AmqpVersion(connectionStart.m_versionMajor,

projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs

@@ -60,7 +60,8 @@ public abstract class ModelBase : IFullModel, IRecoverable
 
         ///<summary>Only used to kick-start a connection open
         ///sequence. See <see cref="Connection.Open"/> </summary>
-        public BlockingCell<ConnectionStartDetails> m_connectionStartCell = null;
+        //public BlockingCell<ConnectionStartDetails> m_connectionStartCell = null;
+        public TaskCompletionSource<ConnectionStartDetails> m_connectionStartCell;
 
         private TimeSpan m_handshakeContinuationTimeout = TimeSpan.FromSeconds(10);
         private TimeSpan m_continuationTimeout = TimeSpan.FromSeconds(20);
@@ -401,7 +402,7 @@ public void FinishClose()
             }
             if (m_connectionStartCell != null)
             {
-                m_connectionStartCell.ContinueWithValue(null);
+                m_connectionStartCell.TrySetResult(null);
             }
         }
 
@@ -958,8 +959,7 @@ public void HandleConnectionStart(byte versionMajor,
                 m_mechanisms = mechanisms,
                 m_locales = locales
             };
-            m_connectionStartCell.ContinueWithValue(details);
-            m_connectionStartCell = null;
+            m_connectionStartCell.TrySetResult(details);
         }
 
         ///<summary>Handle incoming Connection.Tune

projects/client/RabbitMQ.Client/src/client/impl/SimpleBlockingRpcContinuation.cs

@@ -39,7 +39,7 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Diagnostics;
+
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Util;
 
@@ -79,19 +79,19 @@ public virtual Command GetReply(TimeSpan timeout)
             }
         }
 
-        private static void ReportInvalidInvariant(Either<Command,ShutdownEventArgs> result)
+        private static void ReportInvalidInvariant(Either<Command, ShutdownEventArgs> result)
         {
             string error = "Illegal EitherAlternative " + result.Alternative;
         }
 
         public virtual void HandleCommand(Command cmd)
         {
-            m_cell.ContinueWithValue(Either<Command,ShutdownEventArgs>.Left(cmd));
+            m_cell.ContinueWithValue(Either<Command, ShutdownEventArgs>.Left(cmd));
         }
 
         public virtual void HandleModelShutdown(ShutdownEventArgs reason)
         {
-            m_cell.ContinueWithValue(Either<Command,ShutdownEventArgs>.Right(reason));
+            m_cell.ContinueWithValue(Either<Command, ShutdownEventArgs>.Right(reason));
         }
     }
 }