Adding fully asynchronous versions of connect and publish.

Author: stebet

projects/RabbitMQ.Client/client/api/IChannel.cs

@@ -208,6 +208,26 @@ void BasicPublish<TProperties>(string exchange, string routingKey, in TPropertie
         /// </remarks>
         void BasicPublish<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body = default, bool mandatory = false)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader;
+        /// <summary>
+        /// Asynchronously publishes a message.
+        /// </summary>
+        /// <remarks>
+        ///   <para>
+        ///     Routing key must be shorter than 255 bytes.
+        ///   </para>
+        /// </remarks>
+        ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body = default, bool mandatory = false)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader;
+        /// <summary>
+        /// Asynchronously publishes a message.
+        /// </summary>
+        /// <remarks>
+        ///   <para>
+        ///     Routing key must be shorter than 255 bytes.
+        ///   </para>
+        /// </remarks>
+        ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body = default, bool mandatory = false)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader;
 #nullable disable
 
         /// <summary>

projects/RabbitMQ.Client/client/api/IChannelExtensions.cs

@@ -31,6 +31,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using RabbitMQ.Client.client.impl;
 
 namespace RabbitMQ.Client
@@ -93,6 +94,12 @@ public static void BasicPublish(this IChannel channel, string exchange, string r
 
         public static void BasicPublish(this IChannel channel, CachedString exchange, CachedString routingKey, ReadOnlyMemory<byte> body = default, bool mandatory = false)
             => channel.BasicPublish(exchange, routingKey, in EmptyBasicProperty.Empty, body, mandatory);
+        
+        public static ValueTask BasicPublishAsync(this IChannel channel, string exchange, string routingKey, ReadOnlyMemory<byte> body = default, bool mandatory = false)
+            => channel.BasicPublishAsync(exchange, routingKey, in EmptyBasicProperty.Empty, body, mandatory);
+
+        public static ValueTask BasicPublishAsync(this IChannel channel, CachedString exchange, CachedString routingKey, ReadOnlyMemory<byte> body = default, bool mandatory = false)
+            => channel.BasicPublishAsync(exchange, routingKey, in EmptyBasicProperty.Empty, body, mandatory);
 #nullable disable
 
         /// <summary>

projects/RabbitMQ.Client/client/framing/Channel.cs

@@ -30,6 +30,7 @@
 //---------------------------------------------------------------------------
 
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using RabbitMQ.Client.client.framing;
 using RabbitMQ.Client.Impl;
 
@@ -108,6 +109,11 @@ public override void _Private_ConnectionOpen(string virtualHost)
         {
             ChannelSend(new ConnectionOpen(virtualHost));
         }
+        
+        public override ValueTask _Private_ConnectionOpenAsync(string virtualHost)
+        {
+            return ModelSendAsync(new ConnectionOpen(virtualHost));
+        }
 
         public override void _Private_ConnectionSecureOk(byte[] response)
         {

projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs

@@ -317,6 +317,14 @@ public void BasicPublish<TProperties>(string exchange, string routingKey, in TPr
         public void BasicPublish<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
             => InnerChannel.BasicPublish(exchange, routingKey, in basicProperties, body, mandatory);
+        
+        public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader
+            => InnerChannel.BasicPublishAsync(exchange, routingKey, in basicProperties, body, mandatory);
+
+        public ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader
+            => InnerChannel.BasicPublishAsync(exchange, routingKey, in basicProperties, body, mandatory);
 
         public void BasicQos(uint prefetchSize, ushort prefetchCount, bool global)
         {

projects/RabbitMQ.Client/client/impl/ChannelBase.cs

@@ -49,9 +49,11 @@ namespace RabbitMQ.Client.Impl
     internal abstract class ChannelBase : IChannel, IRecoverable
     {
         ///<summary>Only used to kick-start a connection open
-        ///sequence. See <see cref="Connection.Open"/> </summary>
-        internal BlockingCell<ConnectionStartDetails> m_connectionStartCell;
+        ///sequence. See <see cref="Connection.OpenAsync"/> </summary>
+        internal TaskCompletionSource<ConnectionStartDetails> m_connectionStartCell;
 
+        // AMQP only allows one RPC operation to be active at a time. 
+        private readonly SemaphoreSlim _rpcSemaphore = new SemaphoreSlim(1, 1);
         private readonly RpcContinuationQueue _continuationQueue = new RpcContinuationQueue();
         private readonly ManualResetEventSlim _flowControlBlock = new ManualResetEventSlim(true);
 
@@ -258,13 +260,19 @@ internal void ConnectionOpen(string virtualHost)
                 k.GetReply(HandshakeContinuationTimeout);
             }
         }
+        
+        internal ValueTask ConnectionOpenAsync(string virtualHost)
+        {
+            return _Private_ConnectionOpenAsync(virtualHost);
+        }
 
-        internal ConnectionSecureOrTune ConnectionSecureOk(byte[] response)
+        internal async ValueTask<ConnectionSecureOrTune> ConnectionSecureOkAsync(byte[] response)
         {
-            var k = new ConnectionStartRpcContinuation();
-            lock (_rpcLock)
+            var k = new ConnectionSecureOrTuneContinuation();
+            await _rpcSemaphore.WaitAsync().ConfigureAwait(false);
+            Enqueue(k);
+            try
             {
-                Enqueue(k);
                 try
                 {
                     _Private_ConnectionSecureOk(response);
@@ -275,31 +283,40 @@ internal ConnectionSecureOrTune ConnectionSecureOk(byte[] response)
                     // which is a much more suitable exception before connection
                     // negotiation finishes
                 }
-                k.GetReply(HandshakeContinuationTimeout);
+
+                return await k;
+            }
+            finally
+            {
+                _rpcSemaphore.Release();
             }
-            return k.m_result;
         }
 
-        internal ConnectionSecureOrTune ConnectionStartOk(IDictionary<string, object> clientProperties, string mechanism, byte[] response, string locale)
+        internal async ValueTask<ConnectionSecureOrTune> ConnectionStartOkAsync(IDictionary<string, object> clientProperties, string mechanism, byte[] response,
+            string locale)
         {
-            var k = new ConnectionStartRpcContinuation();
-            lock (_rpcLock)
+            var k = new ConnectionSecureOrTuneContinuation();
+            await _rpcSemaphore.WaitAsync().ConfigureAwait(false);
+            Enqueue(k);
+            try
             {
-                Enqueue(k);
                 try
                 {
-                    _Private_ConnectionStartOk(clientProperties, mechanism,
-                        response, locale);
+                    _Private_ConnectionStartOk(clientProperties, mechanism, response, locale);
                 }
                 catch (AlreadyClosedException)
                 {
                     // let continuation throw OperationInterruptedException,
                     // which is a much more suitable exception before connection
                     // negotiation finishes
                 }
-                k.GetReply(HandshakeContinuationTimeout);
+
+                return await k;
+            }
+            finally
+            {
+                _rpcSemaphore.Release();
             }
-            return k.m_result;
         }
 
         protected abstract bool DispatchAsynchronous(in IncomingCommand cmd);
@@ -324,7 +341,7 @@ internal void FinishClose()
                 Session.Close(reason);
             }
 
-            m_connectionStartCell?.ContinueWithValue(null);
+            m_connectionStartCell?.TrySetResult(null);
         }
 
         private void HandleCommand(in IncomingCommand cmd)
@@ -384,6 +401,12 @@ protected void ChannelSend<T>(in T method) where T : struct, IOutgoingAmqpMethod
         {
             Session.Transmit(in method);
         }
+        
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        protected ValueTask ModelSendAsync<T>(in T method) where T : struct, IOutgoingAmqpMethod
+        {
+            return Session.TransmitAsync(in method);
+        }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         protected void ChannelSend<TMethod, THeader>(in TMethod method, in THeader header, ReadOnlyMemory<byte> body)
@@ -396,6 +419,19 @@ protected void ChannelSend<TMethod, THeader>(in TMethod method, in THeader heade
             }
             Session.Transmit(in method, in header, body);
         }
+        
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        protected ValueTask ModelSendAsync<TMethod, THeader>(in TMethod method, in THeader header, ReadOnlyMemory<byte> body)
+            where TMethod : struct, IOutgoingAmqpMethod
+            where THeader : IAmqpHeader
+        {
+            if (!_flowControlBlock.IsSet)
+            {
+                _flowControlBlock.Wait();
+            }
+            
+            return Session.TransmitAsync(in method, in header, body);
+        }
 
         internal void OnCallbackException(CallbackExceptionEventArgs args)
         {
@@ -730,13 +766,7 @@ protected void HandleConnectionClose(in IncomingCommand cmd)
 
         protected void HandleConnectionSecure(in IncomingCommand cmd)
         {
-            var challenge = new ConnectionSecure(cmd.MethodBytes.Span)._challenge;
-            cmd.ReturnMethodBuffer();
-            var k = (ConnectionStartRpcContinuation)_continuationQueue.Next();
-            k.m_result = new ConnectionSecureOrTune
-            {
-                m_challenge = challenge
-            };
+            var k = (ConnectionSecureOrTuneContinuation)_continuationQueue.Next();
             k.HandleCommand(IncomingCommand.Empty); // release the continuation.
         }
 
@@ -758,25 +788,14 @@ protected void HandleConnectionStart(in IncomingCommand cmd)
                 m_mechanisms = method._mechanisms,
                 m_locales = method._locales
             };
-            m_connectionStartCell.ContinueWithValue(details);
+            m_connectionStartCell?.SetResult(details);
             m_connectionStartCell = null;
         }
 
         protected void HandleConnectionTune(in IncomingCommand cmd)
         {
-            var connectionTune = new ConnectionTune(cmd.MethodBytes.Span);
-            cmd.ReturnMethodBuffer();
-            var k = (ConnectionStartRpcContinuation)_continuationQueue.Next();
-            k.m_result = new ConnectionSecureOrTune
-            {
-                m_tuneDetails =
-                {
-                    m_channelMax = connectionTune._channelMax,
-                    m_frameMax = connectionTune._frameMax,
-                    m_heartbeatInSeconds = connectionTune._heartbeat
-                }
-            };
-            k.HandleCommand(IncomingCommand.Empty); // release the continuation.
+            var k = (ConnectionSecureOrTuneContinuation)_continuationQueue.Next();
+            k.HandleCommand(cmd); // release the continuation.
         }
 
         protected void HandleConnectionUnblocked()
@@ -815,6 +834,8 @@ protected void HandleQueueDeclareOk(in IncomingCommand cmd)
 
         public abstract void _Private_ConnectionOpen(string virtualHost);
 
+        public abstract ValueTask _Private_ConnectionOpenAsync(string virtualHost);
+
         public abstract void _Private_ConnectionSecureOk(byte[] response);
 
         public abstract void _Private_ConnectionStartOk(IDictionary<string, object> clientProperties, string mechanism, byte[] response, string locale);
@@ -929,6 +950,36 @@ public void BasicPublish<TProperties>(CachedString exchange, CachedString routin
             var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);
             ChannelSend(in cmd, in basicProperties, body);
         }
+        
+        public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader
+        {
+            if (NextPublishSeqNo > 0)
+            {
+                lock (_confirmLock)
+                {
+                    _pendingDeliveryTags.AddLast(NextPublishSeqNo++);
+                }
+            }
+
+            var cmd = new BasicPublish(exchange, routingKey, mandatory, default);
+            return ModelSendAsync(in cmd, in basicProperties, body);
+        }
+        
+        public ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader
+        {
+            if (NextPublishSeqNo > 0)
+            {
+                lock (_confirmLock)
+                {
+                    _pendingDeliveryTags.AddLast(NextPublishSeqNo++);
+                }
+            }
+
+            var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);
+            return ModelSendAsync(in cmd, in basicProperties, body);
+        }
 
         public void UpdateSecret(string newSecret, string reason)
         {

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

@@ -32,6 +32,7 @@
 using System;
 using System.IO;
 using System.Text;
+using System.Threading.Tasks;
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Impl;
@@ -70,22 +71,21 @@ internal void HandleConnectionUnblocked()
             }
         }
 
-        private void Open()
+        private async ValueTask OpenAsync()
         {
             RabbitMqClientEventSource.Log.ConnectionOpened();
-            StartAndTune();
-            _channel0.ConnectionOpen(_config.VirtualHost);
+            await StartAndTune().ConfigureAwait(false);
+            await _channel0.ConnectionOpenAsync(_config.VirtualHost);
         }
 
-        private void StartAndTune()
+        private async ValueTask StartAndTune()
         {
-            var connectionStartCell = new BlockingCell<ConnectionStartDetails>();
+            var connectionStartCell = new TaskCompletionSource<ConnectionStartDetails>(TaskCreationOptions.RunContinuationsAsynchronously);
             _channel0.m_connectionStartCell = connectionStartCell;
             _channel0.HandshakeContinuationTimeout = _config.HandshakeContinuationTimeout;
             _frameHandler.ReadTimeout = _config.HandshakeContinuationTimeout;
-            _frameHandler.SendHeader();
-
-            ConnectionStartDetails connectionStart = connectionStartCell.WaitForValue();
+            await _frameHandler.SendHeader().ConfigureAwait(false);
+            ConnectionStartDetails connectionStart = await connectionStartCell.Task.ConfigureAwait(false);
 
             if (connectionStart is null)
             {
@@ -117,14 +117,14 @@ private void StartAndTune()
                     ConnectionSecureOrTune res;
                     if (challenge is null)
                     {
-                        res = _channel0.ConnectionStartOk(ClientProperties,
+                        res = await _channel0.ConnectionStartOkAsync(ClientProperties,
                             mechanismFactory.Name,
                             response,
-                            "en_US");
+                            "en_US").ConfigureAwait(false);
                     }
                     else
                     {
-                        res = _channel0.ConnectionSecureOk(response);
+                        res = await _channel0.ConnectionSecureOkAsync(response).ConfigureAwait(false);
                     }
 
                     if (res.m_challenge is null)

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

@@ -82,7 +82,7 @@ public Connection(ConnectionConfig config, IFrameHandler frameHandler)
             _mainLoopTask = Task.Run(MainLoop);
             try
             {
-                Open();
+                OpenAsync().AsTask().GetAwaiter().GetResult();
             }
             catch
             {
@@ -402,7 +402,16 @@ internal void OnCallbackException(CallbackExceptionEventArgs args)
 
         internal void Write(ReadOnlyMemory<byte> memory)
         {
-            _frameHandler.Write(memory);
+            var task = _frameHandler.WriteAsync(memory);
+            if (!task.IsCompletedSuccessfully)
+            {
+                task.AsTask().GetAwaiter().GetResult();
+            }
+        }
+        
+        internal ValueTask WriteAsync(ReadOnlyMemory<byte> memory)
+        {
+            return _frameHandler.WriteAsync(memory);
         }
 
         public void Dispose()

projects/RabbitMQ.Client/client/impl/IFrameHandler.cs

@@ -65,8 +65,8 @@ internal interface IFrameHandler
         ///</summary>
         bool TryReadFrame(out InboundFrame frame);
 
-        void SendHeader();
+        ValueTask SendHeader();
 
-        void Write(ReadOnlyMemory<byte> memory);
+        ValueTask WriteAsync(ReadOnlyMemory<byte> memory);
     }
 }