Separated publishing into separate channel and task queue

Hopefully fixes #23

07-ParallelizationTest/Program.cs

@@ -72,14 +72,16 @@ namespace _07_ParallelizationTest
 
 
             var publisher = dependencyResolver.Resolve<IPublisher>();
-            Console.WriteLine($"Publishing {MessageCount * RepeatBatch} messages...");
+            Console.WriteLine($"Publishing first {MessageCount} of {MessageCount * RepeatBatch} messages...");
 
-            await PublishMessages(publisher, MessageCount * RepeatBatch);
+            await PublishMessages(publisher, MessageCount);
 
 
-
+            Console.WriteLine("Consuming messages while publishing the rest...");
-            Console.WriteLine("Consuming messages...");
             await subscriber.Resume();
+
+            await PublishMessages(publisher, MessageCount * (RepeatBatch - 1));
+
             await waitForDone();
         }
 

Examples/03-FlowRequestResponse/SimpleFlowController.cs

@@ -32,6 +32,12 @@ namespace _03_FlowRequestResponse
         public DateTime RequestStartTime;
 
         
+        // Be sure not to accidentally use any public fields that aren't serializable, for example:
+        //public TaskCompletionSource<bool> SerializationFail = new TaskCompletionSource<bool>();
+        //
+        // In the Newtonsoft.Json version at the time of writing, this will not result in an exception but instead hang the flow!
+
+
         public SimpleFlowController(IFlowProvider flowProvider, IExampleState exampleState)
         {
             this.flowProvider = flowProvider;

Tapeti.Cmd/Program.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.ComponentModel;
 using System.Diagnostics;
 using System.IO;
 using System.Linq;

Tapeti/Connection/TapetiChannel.cs

@@ -0,0 +1,125 @@
+using System;
+using System.Threading.Tasks;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Exceptions;
+using Tapeti.Tasks;
+
+namespace Tapeti.Connection
+{
+    internal interface ITapetiChannelModelProvider
+    {
+        void WithChannel(Action<IModel> operation);
+        void WithRetryableChannel(Action<IModel> operation);
+    }
+    
+    
+    /// <summary>
+    /// Represents both a RabbitMQ Client Channel (IModel) as well as it's associated single-thread task queue.
+    /// Access to the IModel is limited by design to enforce this relationship.
+    /// </summary>
+    internal class TapetiChannel
+    {
+        private readonly Func<IModel> modelFactory;
+        private readonly object taskQueueLock = new();
+        private SingleThreadTaskQueue taskQueue;
+        private readonly ModelProvider modelProvider;
+
+
+        public TapetiChannel(Func<IModel> modelFactory)
+        {
+            this.modelFactory = modelFactory;
+            modelProvider = new ModelProvider(this);
+        }
+
+
+        public async Task Reset()
+        {
+            SingleThreadTaskQueue capturedTaskQueue;
+            
+            lock (taskQueueLock)
+            {
+                capturedTaskQueue = taskQueue;
+                taskQueue = null;
+            }
+
+            if (capturedTaskQueue == null)
+                return;
+            
+            await capturedTaskQueue.Add(() => { });
+            capturedTaskQueue.Dispose();
+        }
+
+
+        public Task Queue(Action<IModel> operation)
+        {
+            return GetTaskQueue().Add(() =>
+            {
+                modelProvider.WithChannel(operation);
+            });
+        }
+
+
+
+        public Task QueueRetryable(Action<IModel> operation)
+        {
+            return GetTaskQueue().Add(() =>
+            {
+                modelProvider.WithRetryableChannel(operation);
+            });
+        }
+
+
+
+        public Task QueueWithProvider(Func<ITapetiChannelModelProvider, Task> operation)
+        {
+            return GetTaskQueue().Add(async () =>
+            {
+                await operation(modelProvider);
+            });
+        }
+
+
+
+        private SingleThreadTaskQueue GetTaskQueue()
+        {
+            lock (taskQueueLock)
+            {
+                return taskQueue ??= new SingleThreadTaskQueue();
+            }
+        }
+
+
+        private class ModelProvider : ITapetiChannelModelProvider
+        {
+            private readonly TapetiChannel owner;
+
+            
+            public ModelProvider(TapetiChannel owner)
+            {
+                this.owner = owner;
+            }
+
+            
+            public void WithChannel(Action<IModel> operation)
+            {
+                operation(owner.modelFactory());
+            }
+            
+
+            public void WithRetryableChannel(Action<IModel> operation)
+            {
+                while (true)
+                {
+                    try
+                    {
+                        operation(owner.modelFactory());
+                        break;
+                    }
+                    catch (AlreadyClosedException)
+                    {
+                    }
+                }
+            }
+        }
+    }
+}

Tapeti/Connection/TapetiClient.cs

@@ -13,10 +13,16 @@ using RabbitMQ.Client.Exceptions;
 using Tapeti.Config;
 using Tapeti.Default;
 using Tapeti.Exceptions;
-using Tapeti.Tasks;
 
 namespace Tapeti.Connection
 {
+    internal enum TapetiChannelType
+    {
+        Consume,
+        Publish
+    }
+    
+    
     /// <inheritdoc />
     /// <summary>
     /// Implementation of ITapetiClient for the RabbitMQ Client library
@@ -39,23 +45,27 @@ namespace Tapeti.Connection
         public IConnectionEventListener ConnectionEventListener { get; set; }
 
 
-        private readonly Lazy<SingleThreadTaskQueue> taskQueue = new Lazy<SingleThreadTaskQueue>();
+        private readonly TapetiChannel consumeChannel;
+        private readonly TapetiChannel publishChannel;
+        private readonly HttpClient managementClient;
 
-        
+        // These fields must be locked using connectionLock
-        // These fields are for use in the taskQueue only!
+        private readonly object connectionLock = new();
         private RabbitMQ.Client.IConnection connection;
+        private IModel consumeChannelModel;
+        private IModel publishChannelModel;
         private bool isClosing;
         private bool isReconnect;
-        private IModel channelInstance;
-        private ulong lastDeliveryTag;
         private DateTime connectedDateTime;
-        private readonly HttpClient managementClient;
-        private readonly HashSet<string> deletedQueues = new HashSet<string>();
 
-        // These fields must be locked, since the callbacks for BasicAck/BasicReturn can run in a different thread
+        // These fields are for use in a single TapetiChannel's queue only!
-        private readonly object confirmLock = new object();
+        private ulong lastDeliveryTag;
-        private readonly Dictionary<ulong, ConfirmMessageInfo> confirmMessages = new Dictionary<ulong, ConfirmMessageInfo>();
+        private readonly HashSet<string> deletedQueues = new();
-        private readonly Dictionary<string, ReturnInfo> returnRoutingKeys = new Dictionary<string, ReturnInfo>();
+
+        // These fields must be locked using confirmLock, since the callbacks for BasicAck/BasicReturn can run in a different thread
+        private readonly object confirmLock = new();
+        private readonly Dictionary<ulong, ConfirmMessageInfo> confirmMessages = new();
+        private readonly Dictionary<string, ReturnInfo> returnRoutingKeys = new();
 
 
         private class ConfirmMessageInfo
@@ -79,6 +89,9 @@ namespace Tapeti.Connection
 
             logger = config.DependencyResolver.Resolve<ILogger>();
 
+            consumeChannel = new TapetiChannel(() => GetModel(TapetiChannelType.Consume));
+            publishChannel = new TapetiChannel(() => GetModel(TapetiChannelType.Publish));
+
 
             var handler = new HttpClientHandler
             {
@@ -100,7 +113,8 @@ namespace Tapeti.Connection
             if (string.IsNullOrEmpty(routingKey))
                 throw new ArgumentNullException(nameof(routingKey));
 
-            await taskQueue.Value.Add(async () =>
+
+            await GetTapetiChannel(TapetiChannelType.Publish).QueueWithProvider(async channelProvider =>
             {
                 Task<int> publishResultTask = null;
                 var messageInfo = new ConfirmMessageInfo
@@ -110,7 +124,7 @@ namespace Tapeti.Connection
                 };
 
 
-                WithRetryableChannel(channel =>
+                channelProvider.WithRetryableChannel(channel =>
                 {
                     DeclareExchange(channel, exchange);
 
@@ -169,15 +183,18 @@ namespace Tapeti.Connection
 
                 var replyCode = publishResultTask.Result;
 
-                // There is no RabbitMQ.Client.Framing.Constants value for this "No route" reply code
+                switch (replyCode)
-                // at the time of writing...
+                {
-                if (replyCode == 312)
+                    // There is no RabbitMQ.Client.Framing.Constants value for this "No route" reply code
-                    throw new NoRouteException(
+                    // at the time of writing...
-                        $"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' does not have a route");
+                    case 312:
+                        throw new NoRouteException(
+                            $"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' does not have a route");
                     
-                if (replyCode > 0)
+                    case > 0:
-                    throw new NoRouteException(
+                        throw new NoRouteException(
-                        $"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' could not be delivered, reply code: {replyCode}");
+                            $"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' could not be delivered, reply code: {replyCode}");
+                }
             });
         }
 
@@ -194,7 +211,7 @@ namespace Tapeti.Connection
 
             string consumerTag = null;
 
-            await QueueWithRetryableChannel(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).QueueRetryable(channel =>
             {
                 if (cancellationToken.IsCancellationRequested)
                     return;
@@ -215,7 +232,7 @@ namespace Tapeti.Connection
 
             // No need for a retryable channel here, if the connection is lost
             // so is the consumer.
-            await Queue(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
             {
                 channel.BasicCancel(consumerTag);
             });
@@ -224,7 +241,7 @@ namespace Tapeti.Connection
 
         private async Task Respond(ulong deliveryTag, ConsumeResult result)
         {
-            await Queue(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
             {
                 // No need for a retryable channel here, if the connection is lost we can't
                 // use the deliveryTag anymore.
@@ -257,7 +274,7 @@ namespace Tapeti.Connection
             var currentBindings = bindings.ToList();
             var bindingLogger = logger as IBindingLogger;
 
-            await Queue(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
             {
                 if (cancellationToken.IsCancellationRequested)
                     return;
@@ -284,7 +301,7 @@ namespace Tapeti.Connection
         /// <inheritdoc />
         public async Task DurableQueueVerify(CancellationToken cancellationToken, string queueName)
         {
-            await Queue(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
             {
                 if (cancellationToken.IsCancellationRequested)
                     return;
@@ -302,7 +319,7 @@ namespace Tapeti.Connection
             {
                 uint deletedMessages = 0;
 
-                await Queue(channel =>
+                await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
                 {
                     if (cancellationToken.IsCancellationRequested)
                         return;
@@ -316,7 +333,7 @@ namespace Tapeti.Connection
             }
 
 
-            await taskQueue.Value.Add(async () =>
+            await GetTapetiChannel(TapetiChannelType.Consume).QueueWithProvider(async channelProvider =>
             {
                 bool retry;
                 do
@@ -343,8 +360,10 @@ namespace Tapeti.Connection
                         // includes the GetQueueInfo, the next time around it should have Messages > 0
                         try
                         {
-                            var channel = GetChannel();
+                            channelProvider.WithChannel(channel =>
-                            channel.QueueDelete(queueName, false, true);
+                            {
+                                channel.QueueDelete(queueName, false, true);
+                            });
 
                             deletedQueues.Add(queueName);
                             (logger as IBindingLogger)?.QueueObsolete(queueName, true, 0);
@@ -364,10 +383,11 @@ namespace Tapeti.Connection
 
                         if (existingBindings.Count > 0)
                         {
-                            var channel = GetChannel();
+                            channelProvider.WithChannel(channel =>
-
+                            {
-                            foreach (var binding in existingBindings)
+                                foreach (var binding in existingBindings)
-                                channel.QueueUnbind(queueName, binding.Exchange, binding.RoutingKey);
+                                    channel.QueueUnbind(queueName, binding.Exchange, binding.RoutingKey);
+                            });
                         }
 
                         (logger as IBindingLogger)?.QueueObsolete(queueName, false, queueInfo.Messages);
@@ -383,7 +403,7 @@ namespace Tapeti.Connection
             string queueName = null;
             var bindingLogger = logger as IBindingLogger;
 
-            await Queue(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
             {
                 if (cancellationToken.IsCancellationRequested)
                     return;
@@ -407,7 +427,7 @@ namespace Tapeti.Connection
         /// <inheritdoc />
         public async Task DynamicQueueBind(CancellationToken cancellationToken, string queueName, QueueBinding binding)
         {
-            await Queue(channel =>
+            await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
             {
                 if (cancellationToken.IsCancellationRequested)
                     return;
@@ -422,32 +442,46 @@ namespace Tapeti.Connection
         /// <inheritdoc />
         public async Task Close()
         {
-            if (!taskQueue.IsValueCreated)
+            IModel capturedConsumeModel;
-                return;
+            IModel capturedPublishModel;
+            RabbitMQ.Client.IConnection capturedConnection;
 
-            await taskQueue.Value.Add(() =>
+            lock (connectionLock)
             {
                 isClosing = true;
+                capturedConsumeModel = consumeChannelModel;
+                capturedPublishModel = publishChannelModel;
+                capturedConnection = connection;
 
-                if (channelInstance != null)
+                consumeChannelModel = null;
+                publishChannelModel = null;
+                connection = null;
+            }
+
+            // Empty the queue
+            await consumeChannel.Reset();
+            await publishChannel.Reset();
+
+            // No need to close the channels as the connection will be closed
+            capturedConsumeModel.Dispose();
+            capturedPublishModel.Dispose();
+
+            // ReSharper disable once InvertIf
+            if (capturedConnection != null)
+            {
+                try
                 {
-                    channelInstance.Dispose();
+                    capturedConnection.Close();
-                    channelInstance = null;
                 }
-
+                finally
-                // ReSharper disable once InvertIf
-                if (connection != null)
                 {
-                    connection.Dispose();
+                    capturedConnection.Dispose();
-                    connection = null;
                 }
-
+            }
-                taskQueue.Value.Dispose();
-            });
         }
 
 
-        private static readonly List<HttpStatusCode> TransientStatusCodes = new List<HttpStatusCode>
+        private static readonly List<HttpStatusCode> TransientStatusCodes = new()
         {
             HttpStatusCode.GatewayTimeout,
             HttpStatusCode.RequestTimeout,
@@ -545,39 +579,37 @@ namespace Tapeti.Connection
         {
             var requestUri = new Uri($"http://{connectionParams.HostName}:{connectionParams.ManagementPort}/api/{path}");
 
-            using (var request = new HttpRequestMessage(HttpMethod.Get, requestUri))
+            using var request = new HttpRequestMessage(HttpMethod.Get, requestUri);
+            var retryDelayIndex = 0;
+
+            while (true)
             {
-                var retryDelayIndex = 0;
+                try
-
-                while (true)
                 {
-                    try
+                    var response = await managementClient.SendAsync(request);
-                    {
+                    return await handleResponse(response);
-                        var response = await managementClient.SendAsync(request);
-                        return await handleResponse(response);
-                    }
-                    catch (TimeoutException)
-                    {
-                    }
-                    catch (WebException e)
-                    {
-                        if (!(e.Response is HttpWebResponse response))
-                            throw;
-
-                        if (!TransientStatusCodes.Contains(response.StatusCode))
-                            throw;
-                    }
-
-                    await Task.Delay(ExponentialBackoff[retryDelayIndex]);
-
-                    if (retryDelayIndex < ExponentialBackoff.Length - 1)
-                        retryDelayIndex++;
                 }
+                catch (TimeoutException)
+                {
+                }
+                catch (WebException e)
+                {
+                    if (!(e.Response is HttpWebResponse response))
+                        throw;
+
+                    if (!TransientStatusCodes.Contains(response.StatusCode))
+                        throw;
+                }
+
+                await Task.Delay(ExponentialBackoff[retryDelayIndex]);
+
+                if (retryDelayIndex < ExponentialBackoff.Length - 1)
+                    retryDelayIndex++;
             }
         }
 
 
-        private readonly HashSet<string> declaredExchanges = new HashSet<string>();
+        private readonly HashSet<string> declaredExchanges = new();
 
         private void DeclareExchange(IModel channel, string exchange)
         {
@@ -593,22 +625,11 @@ namespace Tapeti.Connection
         }
 
 
-        private async Task Queue(Action<IModel> operation)
+        private TapetiChannel GetTapetiChannel(TapetiChannelType channelType)
         {
-            await taskQueue.Value.Add(() =>
+            return channelType == TapetiChannelType.Publish
-            {
+                ? publishChannel
-                var channel = GetChannel();
+                : consumeChannel;
-                operation(channel);
-            });
-        }
-
-
-        private async Task QueueWithRetryableChannel(Action<IModel> operation)
-        {
-            await taskQueue.Value.Add(() =>
-            {
-                WithRetryableChannel(operation);
-            });
         }
 
         
@@ -616,140 +637,170 @@ namespace Tapeti.Connection
         /// Only call this from a task in the taskQueue to ensure IModel is only used 
         /// by a single thread, as is recommended in the RabbitMQ .NET Client documentation.
         /// </remarks>
-        private void WithRetryableChannel(Action<IModel> operation)
+        private IModel GetModel(TapetiChannelType channelType)
         {
-            while (true)
+            lock (connectionLock)
             {
-                try
+                var channel = channelType == TapetiChannelType.Publish
+                    ? publishChannelModel
+                    : consumeChannelModel;
+
+                if (channel != null && channel.IsOpen)
+                    return channel;
+
+                // If the Disconnect quickly follows the Connect (when an error occurs that is reported back by RabbitMQ
+                // not related to the connection), wait for a bit to avoid spamming the connection
+                if ((DateTime.UtcNow - connectedDateTime).TotalMilliseconds <= MinimumConnectedReconnectDelay)
+                    Thread.Sleep(ReconnectDelay);
+
+
+                var connectionFactory = new ConnectionFactory
                 {
-                    operation(GetChannel());
+                    HostName = connectionParams.HostName,
-                    break;
+                    Port = connectionParams.Port,
-                }
+                    VirtualHost = connectionParams.VirtualHost,
-                catch (AlreadyClosedException)
+                    UserName = connectionParams.Username,
-                {
+                    Password = connectionParams.Password,
-                }
+                    AutomaticRecoveryEnabled = false,
-            }
+                    TopologyRecoveryEnabled = false,
-        }
+                    RequestedHeartbeat = TimeSpan.FromSeconds(30)
+                };
 
-
+                if (connectionParams.ClientProperties != null)
-        /// <remarks>
+                    foreach (var pair in connectionParams.ClientProperties)
-        /// Only call this from a task in the taskQueue to ensure IModel is only used 
-        /// by a single thread, as is recommended in the RabbitMQ .NET Client documentation.
-        /// </remarks>
-        private IModel GetChannel()
-        {
-            if (channelInstance != null && channelInstance.IsOpen)
-                return channelInstance;
-
-            // If the Disconnect quickly follows the Connect (when an error occurs that is reported back by RabbitMQ
-            // not related to the connection), wait for a bit to avoid spamming the connection
-            if ((DateTime.UtcNow - connectedDateTime).TotalMilliseconds <= MinimumConnectedReconnectDelay)
-                Thread.Sleep(ReconnectDelay);
-
-
-            var connectionFactory = new ConnectionFactory
-            {
-                HostName = connectionParams.HostName,
-                Port = connectionParams.Port,
-                VirtualHost = connectionParams.VirtualHost,
-                UserName = connectionParams.Username,
-                Password = connectionParams.Password,
-                AutomaticRecoveryEnabled = false,
-                TopologyRecoveryEnabled = false,
-                RequestedHeartbeat = TimeSpan.FromSeconds(30)
-            };
-
-            if (connectionParams.ClientProperties != null)
-                foreach (var pair in connectionParams.ClientProperties)
-                {
-                    if (connectionFactory.ClientProperties.ContainsKey(pair.Key))
-                        connectionFactory.ClientProperties[pair.Key] = Encoding.UTF8.GetBytes(pair.Value);
-                    else
-                        connectionFactory.ClientProperties.Add(pair.Key, Encoding.UTF8.GetBytes(pair.Value));
-                }
-
-
-            while (true)
-            {
-                try
-                {
-                    logger.Connect(new ConnectContext(connectionParams, isReconnect));
-
-                    connection = connectionFactory.CreateConnection();
-                    channelInstance = connection.CreateModel();
-
-                    if (channelInstance == null)
-                        throw new BrokerUnreachableException(null);
-
-                    if (config.Features.PublisherConfirms)
                     {
-                        lastDeliveryTag = 0;
+                        if (connectionFactory.ClientProperties.ContainsKey(pair.Key))
-
+                            connectionFactory.ClientProperties[pair.Key] = Encoding.UTF8.GetBytes(pair.Value);
-                        Monitor.Enter(confirmLock);
+                        else
-                        try
+                            connectionFactory.ClientProperties.Add(pair.Key, Encoding.UTF8.GetBytes(pair.Value));
-                        {
-                            foreach (var pair in confirmMessages)
-                                pair.Value.CompletionSource.SetCanceled();
-
-                            confirmMessages.Clear();
-                        }
-                        finally
-                        {
-                            Monitor.Exit(confirmLock);
-                        }
-
-                        channelInstance.ConfirmSelect();
                     }
 
-                    if (connectionParams.PrefetchCount > 0)
-                        channelInstance.BasicQos(0, connectionParams.PrefetchCount, false);
 
-                    channelInstance.ModelShutdown += (sender, e) =>
+                while (true)
-                    {
-                        ConnectionEventListener?.Disconnected(new DisconnectedEventArgs
-                        {
-                            ReplyCode = e.ReplyCode,
-                            ReplyText = e.ReplyText
-                        });
-
-                        logger.Disconnect(new DisconnectContext(connectionParams, e.ReplyCode, e.ReplyText));
-
-                        channelInstance = null;
-
-                        if (!isClosing)
-                            taskQueue.Value.Add(() => WithRetryableChannel(channel => { }));
-                    };
-
-                    channelInstance.BasicReturn += HandleBasicReturn;
-                    channelInstance.BasicAcks += HandleBasicAck;
-                    channelInstance.BasicNacks += HandleBasicNack;
-
-                    connectedDateTime = DateTime.UtcNow;
-
-                    var connectedEventArgs = new ConnectedEventArgs
-                    {
-                        ConnectionParams = connectionParams,
-                        LocalPort = connection.LocalPort
-                    };
-
-                    if (isReconnect)
-                        ConnectionEventListener?.Reconnected(connectedEventArgs);
-                    else
-                        ConnectionEventListener?.Connected(connectedEventArgs);
-
-                    logger.ConnectSuccess(new ConnectContext(connectionParams, isReconnect, connection.LocalPort));
-                    isReconnect = true;
-
-                    break;
-                }
-                catch (BrokerUnreachableException e)
                 {
-                    logger.ConnectFailed(new ConnectContext(connectionParams, isReconnect, exception: e));
+                    try
-                    Thread.Sleep(ReconnectDelay);
+                    {
-                }
+                        if (connection != null)
-            }
+                        {
+                            try
+                            {
+                                connection.Close();
+                            }
+                            finally
+                            {
+                                connection.Dispose();
+                            }
 
-            return channelInstance;
+                            connection = null;
+                        }
+
+                        logger.Connect(new ConnectContext(connectionParams, isReconnect));
+
+                        connection = connectionFactory.CreateConnection();
+                        consumeChannelModel = connection.CreateModel();
+                        if (consumeChannel == null)
+                            throw new BrokerUnreachableException(null);
+
+                        publishChannelModel = connection.CreateModel();
+                        if (publishChannel == null)
+                            throw new BrokerUnreachableException(null);
+
+
+                        if (config.Features.PublisherConfirms)
+                        {
+                            lastDeliveryTag = 0;
+
+                            Monitor.Enter(confirmLock);
+                            try
+                            {
+                                foreach (var pair in confirmMessages)
+                                    pair.Value.CompletionSource.SetCanceled();
+
+                                confirmMessages.Clear();
+                            }
+                            finally
+                            {
+                                Monitor.Exit(confirmLock);
+                            }
+
+                            publishChannelModel.ConfirmSelect();
+                        }
+
+                        if (connectionParams.PrefetchCount > 0)
+                            consumeChannelModel.BasicQos(0, connectionParams.PrefetchCount, false);
+
+                        var capturedConsumeChannelModel = consumeChannelModel;
+                        consumeChannelModel.ModelShutdown += (_, e) =>
+                        {
+                            lock (connectionLock)
+                            {
+                                if (consumeChannelModel == null || consumeChannelModel != capturedConsumeChannelModel)
+                                    return;
+
+                                consumeChannelModel = null;
+                            }
+
+                            ConnectionEventListener?.Disconnected(new DisconnectedEventArgs
+                            {
+                                ReplyCode = e.ReplyCode,
+                                ReplyText = e.ReplyText
+                            });
+
+                            logger.Disconnect(new DisconnectContext(connectionParams, e.ReplyCode, e.ReplyText));
+
+                            // Reconnect if the disconnect was unexpected
+                            if (!isClosing)
+                                GetTapetiChannel(TapetiChannelType.Consume).QueueRetryable(_ => { });
+                        };
+
+                        var capturedPublishChannelModel = publishChannelModel;
+                        publishChannelModel.ModelShutdown += (_, _) =>
+                        {
+                            lock (connectionLock)
+                            {
+                                if (publishChannelModel == null || publishChannelModel != capturedPublishChannelModel)
+                                    return;
+
+                                publishChannelModel = null;
+                            }
+
+                            // No need to reconnect, the next Publish will
+                        };
+
+
+                        publishChannelModel.BasicReturn += HandleBasicReturn;
+                        publishChannelModel.BasicAcks += HandleBasicAck;
+                        publishChannelModel.BasicNacks += HandleBasicNack;
+
+                        connectedDateTime = DateTime.UtcNow;
+
+                        var connectedEventArgs = new ConnectedEventArgs
+                        {
+                            ConnectionParams = connectionParams,
+                            LocalPort = connection.LocalPort
+                        };
+
+                        if (isReconnect)
+                            ConnectionEventListener?.Reconnected(connectedEventArgs);
+                        else
+                            ConnectionEventListener?.Connected(connectedEventArgs);
+
+                        logger.ConnectSuccess(new ConnectContext(connectionParams, isReconnect, connection.LocalPort));
+                        isReconnect = true;
+
+                        break;
+                    }
+                    catch (BrokerUnreachableException e)
+                    {
+                        logger.ConnectFailed(new ConnectContext(connectionParams, isReconnect, exception: e));
+                        Thread.Sleep(ReconnectDelay);
+                    }
+                }
+                
+                return channelType == TapetiChannelType.Publish
+                    ? publishChannelModel
+                    : consumeChannelModel;
+            }
         }
 
 

Tapeti/Connection/TapetiConsumer.cs

@@ -57,7 +57,7 @@ namespace Tapeti.Connection
             }
             catch (Exception dispatchException)
             {
-                using (var emptyContext = new MessageContext
+                await using var emptyContext = new MessageContext
                 {
                     Config = config,
                     Queue = queueName,
@@ -66,12 +66,11 @@ namespace Tapeti.Connection
                     Message = message,
                     Properties = properties,
                     Binding = null
-                })
+                };
-                {
+                
-                    var exceptionContext = new ExceptionStrategyContext(emptyContext, dispatchException);
+                var exceptionContext = new ExceptionStrategyContext(emptyContext, dispatchException);
-                    HandleException(exceptionContext);
+                HandleException(exceptionContext);
-                    return exceptionContext.ConsumeResult;
+                return exceptionContext.ConsumeResult;
-                }
             }
         }
 
@@ -100,7 +99,7 @@ namespace Tapeti.Connection
 
         private async Task<ConsumeResult> InvokeUsingBinding(object message, MessageContextData messageContextData, IBinding binding)
         {
-            using (var context = new MessageContext
+            await using var context = new MessageContext
             {
                 Config = config,
                 Queue = queueName,
@@ -109,25 +108,24 @@ namespace Tapeti.Connection
                 Message = message,
                 Properties = messageContextData.Properties,
                 Binding = binding
-            })
+            };
+            
+            try
             {
-                try
+                await MiddlewareHelper.GoAsync(config.Middleware.Message,
-                {
+                    async (handler, next) => await handler.Handle(context, next),
-                    await MiddlewareHelper.GoAsync(config.Middleware.Message,
+                    async () => { await binding.Invoke(context); });
-                        async (handler, next) => await handler.Handle(context, next),
-                        async () => { await binding.Invoke(context); });
 
-                    await binding.Cleanup(context, ConsumeResult.Success);
+                await binding.Cleanup(context, ConsumeResult.Success);
-                    return ConsumeResult.Success;
+                return ConsumeResult.Success;
-                }
+            }
-                catch (Exception invokeException)
+            catch (Exception invokeException)
-                {
+            {
-                    var exceptionContext = new ExceptionStrategyContext(context, invokeException);
+                var exceptionContext = new ExceptionStrategyContext(context, invokeException);
-                    HandleException(exceptionContext);
+                HandleException(exceptionContext);
 
-                    await binding.Cleanup(context, exceptionContext.ConsumeResult);
+                await binding.Cleanup(context, exceptionContext.ConsumeResult);
-                    return exceptionContext.ConsumeResult;
+                return exceptionContext.ConsumeResult;
-                }
             }
         }
 
@@ -158,18 +156,12 @@ namespace Tapeti.Connection
 
         private static bool IgnoreExceptionDuringShutdown(Exception e)
         {
-            switch (e)
+            return e switch
             {
-                case AggregateException aggregateException:
+                AggregateException aggregateException => aggregateException.InnerExceptions.Any(IgnoreExceptionDuringShutdown),
-                    return aggregateException.InnerExceptions.Any(IgnoreExceptionDuringShutdown);
+                TaskCanceledException or OperationCanceledException => true,
-
+                _ => e.InnerException != null && IgnoreExceptionDuringShutdown(e.InnerException)
-                case TaskCanceledException _:
+            };
-                case OperationCanceledException _:  // thrown by CancellationTokenSource.ThrowIfCancellationRequested
-                    return true;
-
-                default:
-                    return e.InnerException != null && IgnoreExceptionDuringShutdown(e.InnerException);
-            }
         }
 
 

Tapeti/Connection/TapetiPublisher.cs

@@ -118,9 +118,7 @@ namespace Tapeti.Connection
         {
             var writableProperties = new MessageProperties(properties);
 
-            if (!writableProperties.Timestamp.HasValue)
+            writableProperties.Timestamp ??= DateTime.UtcNow;
-                writableProperties.Timestamp = DateTime.UtcNow;
-
             writableProperties.Persistent = true;
 
 

Tapeti/Connection/TapetiSubscriber.cs

@@ -13,7 +13,7 @@ namespace Tapeti.Connection
         private readonly Func<ITapetiClient> clientFactory;
         private readonly ITapetiConfig config;
         private bool consuming;
-        private readonly List<string> consumerTags = new List<string>();
+        private readonly List<string> consumerTags = new();
 
         private CancellationTokenSource initializeCancellationTokenSource;
 
@@ -166,7 +166,7 @@ namespace Tapeti.Connection
                 public List<Type> MessageClasses;
             }
 
-            private readonly Dictionary<string, List<DynamicQueueInfo>> dynamicQueues = new Dictionary<string, List<DynamicQueueInfo>>();
+            private readonly Dictionary<string, List<DynamicQueueInfo>> dynamicQueues = new();
 
 
             protected CustomBindingTarget(Func<ITapetiClient> clientFactory, IRoutingKeyStrategy routingKeyStrategy, IExchangeStrategy exchangeStrategy, CancellationToken cancellationToken)
@@ -277,8 +277,8 @@ namespace Tapeti.Connection
 
         private class DeclareDurableQueuesBindingTarget : CustomBindingTarget
         {
-            private readonly Dictionary<string, List<Type>> durableQueues = new Dictionary<string, List<Type>>();
+            private readonly Dictionary<string, List<Type>> durableQueues = new();
-            private readonly HashSet<string> obsoleteDurableQueues = new HashSet<string>();
+            private readonly HashSet<string> obsoleteDurableQueues = new();
 
 
             public DeclareDurableQueuesBindingTarget(Func<ITapetiClient> clientFactory, IRoutingKeyStrategy routingKeyStrategy, IExchangeStrategy exchangeStrategy, CancellationToken cancellationToken) : base(clientFactory, routingKeyStrategy, exchangeStrategy, cancellationToken)
@@ -358,7 +358,7 @@ namespace Tapeti.Connection
 
         private class PassiveDurableQueuesBindingTarget : CustomBindingTarget
         {
-            private readonly List<string> durableQueues = new List<string>();
+            private readonly List<string> durableQueues = new();
 
 
             public PassiveDurableQueuesBindingTarget(Func<ITapetiClient> clientFactory, IRoutingKeyStrategy routingKeyStrategy, IExchangeStrategy exchangeStrategy, CancellationToken cancellationToken) : base(clientFactory, routingKeyStrategy, exchangeStrategy, cancellationToken)

Tapeti/Default/ControllerBindingContext.cs

@@ -9,7 +9,7 @@ namespace Tapeti.Default
     internal class ControllerBindingContext : IControllerBindingContext
     {
         private BindingTargetMode? bindingTargetMode;
-        private readonly List<IControllerMiddlewareBase> middleware = new List<IControllerMiddlewareBase>();
+        private readonly List<IControllerMiddlewareBase> middleware = new();
         private readonly List<ControllerBindingParameter> parameters;
         private readonly ControllerBindingResult result;
 

Tapeti/Default/ControllerMethodBinding.cs

@@ -161,36 +161,34 @@ namespace Tapeti.Default
         {
             var controller = dependencyResolver.Resolve(bindingInfo.ControllerType);
 
-            using (var controllerContext = new ControllerMessageContext(context)
+            await using var controllerContext = new ControllerMessageContext(context)
             {
                 Controller = controller
-            })
+            };
-            {
+            
-                if (!await FilterAllowed(controllerContext))
+            if (!await FilterAllowed(controllerContext))
-                    return;
+                return;
 
 
-                await MiddlewareHelper.GoAsync(
+            await MiddlewareHelper.GoAsync(
-                    bindingInfo.MessageMiddleware,
+                bindingInfo.MessageMiddleware,
-                    async (handler, next) => await handler.Handle(controllerContext, next),
+                async (handler, next) => await handler.Handle(controllerContext, next),
-                    async () => await messageHandler(controllerContext));
+                async () => await messageHandler(controllerContext));
-            }
         }
 
 
         /// <inheritdoc />
         public async Task Cleanup(IMessageContext context, ConsumeResult consumeResult)
         {
-            using (var controllerContext = new ControllerMessageContext(context)
+            await using var controllerContext = new ControllerMessageContext(context)
             {
                 Controller = null
-            })
+            };
-            {
+            
-                await MiddlewareHelper.GoAsync(
+            await MiddlewareHelper.GoAsync(
-                    bindingInfo.CleanupMiddleware,
+                bindingInfo.CleanupMiddleware,
-                    async (handler, next) => await handler.Cleanup(controllerContext, consumeResult, next),
+                async (handler, next) => await handler.Cleanup(controllerContext, consumeResult, next),
-                    () => Task.CompletedTask);
+                () => Task.CompletedTask);
-            }
         }
 
 

Tapeti/Default/JsonMessageSerializer.cs

@@ -16,8 +16,8 @@ namespace Tapeti.Default
         private const string ClassTypeHeader = "classType";
 
 
-        private readonly ConcurrentDictionary<string, Type> deserializedTypeNames = new ConcurrentDictionary<string, Type>();
+        private readonly ConcurrentDictionary<string, Type> deserializedTypeNames = new();
-        private readonly ConcurrentDictionary<Type, string> serializedTypeNames = new ConcurrentDictionary<Type, string>();
+        private readonly ConcurrentDictionary<Type, string> serializedTypeNames = new();
         private readonly JsonSerializerSettings serializerSettings;
 
 

Tapeti/Default/MessageContext.cs

@@ -7,7 +7,7 @@ namespace Tapeti.Default
 {
     internal class MessageContext : IMessageContext
     {
-        private readonly Dictionary<string, object> items = new Dictionary<string, object>();
+        private readonly Dictionary<string, object> items = new();
 
 
         /// <inheritdoc />

Tapeti/Default/MessageProperties.cs

@@ -10,7 +10,7 @@ namespace Tapeti.Default
     /// </summary>
     public class MessageProperties : IMessageProperties
     {
-        private readonly Dictionary<string, string> headers = new Dictionary<string, string>();
+        private readonly Dictionary<string, string> headers = new();
 
 
         /// <inheritdoc />

Tapeti/Default/NamespaceMatchExchangeStrategy.cs

@@ -13,7 +13,7 @@ namespace Tapeti.Default
     /// </example>
     public class NamespaceMatchExchangeStrategy : IExchangeStrategy
     {
-        private static readonly Regex NamespaceRegex = new Regex("^(Messaging\\.)?(?<exchange>[^\\.]+)", RegexOptions.Compiled | RegexOptions.Singleline);
+        private static readonly Regex NamespaceRegex = new("^(Messaging\\.)?(?<exchange>[^\\.]+)", RegexOptions.Compiled | RegexOptions.Singleline);
 
 
         /// <inheritdoc />

Tapeti/Default/RabbitMQMessageProperties.cs

@@ -87,8 +87,7 @@ namespace Tapeti.Default
         /// <inheritdoc />
         public void SetHeader(string name, string value)
         {
-            if (BasicProperties.Headers == null)
+            BasicProperties.Headers ??= new Dictionary<string, object>();
-                BasicProperties.Headers = new Dictionary<string, object>();
 
             if (BasicProperties.Headers.ContainsKey(name))
                 BasicProperties.Headers[name] = Encoding.UTF8.GetBytes(value);

Tapeti/Default/TypeNameRoutingKeyStrategy.cs

@@ -28,9 +28,9 @@ namespace Tapeti.Default
                     (?(?<=[A-Z])[A-Z](?=[a-z])|[A-Z])
                 )";
 
-        private static readonly Regex SeparatorRegex = new Regex(SeparatorPattern, RegexOptions.IgnorePatternWhitespace | RegexOptions.Compiled);
+        private static readonly Regex SeparatorRegex = new(SeparatorPattern, RegexOptions.IgnorePatternWhitespace | RegexOptions.Compiled);
 
-        private static readonly ConcurrentDictionary<Type, string> RoutingKeyCache = new ConcurrentDictionary<Type, string>();
+        private static readonly ConcurrentDictionary<Type, string> RoutingKeyCache = new();
 
 
         /// <inheritdoc />

Tapeti/Helpers/ConnectionstringParser.cs

@@ -8,7 +8,7 @@ namespace Tapeti.Helpers
     /// </summary>
     public class ConnectionStringParser
     {
-        private readonly TapetiConnectionParams result = new TapetiConnectionParams();
+        private readonly TapetiConnectionParams result = new();
 
         private readonly string connectionstring;
         private int pos = -1;

Tapeti/ISubscriber.cs

@@ -2,6 +2,7 @@
 using System.Threading.Tasks;
 
 // ReSharper disable UnusedMember.Global
+// ReSharper disable UnusedMemberInSuper.Global
 
 namespace Tapeti
 {

Tapeti/Tapeti.csproj

@@ -11,6 +11,7 @@
     <PackageLicenseExpression>Unlicense</PackageLicenseExpression>
     <PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
     <PackageIcon>Tapeti.png</PackageIcon>
+    <LangVersion>latest</LangVersion>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">

Tapeti/TapetiConfig.cs

@@ -18,7 +18,7 @@ namespace Tapeti
     public class TapetiConfig : ITapetiConfigBuilder, ITapetiConfigBuilderAccess
     {
         private Config config;
-        private readonly List<IControllerBindingMiddleware> bindingMiddleware = new List<IControllerBindingMiddleware>();
+        private readonly List<IControllerBindingMiddleware> bindingMiddleware = new();
 
 
         /// <inheritdoc />
@@ -112,7 +112,7 @@ namespace Tapeti
 
                         default:
                             throw new ArgumentException(
-                                $"Unsupported middleware implementation: {(middleware == null ? "null" : middleware.GetType().Name)}");
+                                $"Unsupported middleware implementation: {middleware?.GetType().Name ?? "null"}");
                     }
                 }
             }
@@ -224,9 +224,9 @@ namespace Tapeti
         /// <inheritdoc />
         internal class Config : ITapetiConfig
         {
-            private readonly ConfigFeatures features = new ConfigFeatures();
+            private readonly ConfigFeatures features = new();
-            private readonly ConfigMiddleware middleware = new ConfigMiddleware();
+            private readonly ConfigMiddleware middleware = new();
-            private readonly ConfigBindings bindings = new ConfigBindings();
+            private readonly ConfigBindings bindings = new();
 
             public IDependencyResolver DependencyResolver { get; }
             public ITapetiConfigFeatues Features => features;
@@ -290,8 +290,8 @@ namespace Tapeti
 
         internal class ConfigMiddleware : ITapetiConfigMiddleware
         {
-            private readonly List<IMessageMiddleware> messageMiddleware = new List<IMessageMiddleware>();
+            private readonly List<IMessageMiddleware> messageMiddleware = new();
-            private readonly List<IPublishMiddleware> publishMiddleware = new List<IPublishMiddleware>();
+            private readonly List<IPublishMiddleware> publishMiddleware = new();
 
 
             public IReadOnlyList<IMessageMiddleware> Message => messageMiddleware;

Tapeti/TapetiConnectionParams.cs

@@ -60,7 +60,7 @@ namespace Tapeti
         /// will be overwritten. See DefaultClientProperties in Connection.cs in the RabbitMQ .NET client source for the default values.
         /// </remarks>
         public IDictionary<string, string> ClientProperties { 
-            get => clientProperties ?? (clientProperties = new Dictionary<string, string>());
+            get => clientProperties ??= new Dictionary<string, string>();
             set => clientProperties = value;
         }
 

Tapeti/Tasks/SingleThreadTaskQueue.cs

@@ -12,10 +12,10 @@ namespace Tapeti.Tasks
     /// </summary>
     public class SingleThreadTaskQueue : IDisposable
     {
-        private readonly object previousTaskLock = new object();
+        private readonly object previousTaskLock = new();
         private Task previousTask = Task.CompletedTask;
 
-        private readonly Lazy<SingleThreadTaskScheduler> singleThreadScheduler = new Lazy<SingleThreadTaskScheduler>();
+        private readonly Lazy<SingleThreadTaskScheduler> singleThreadScheduler = new();
 
 
         /// <summary>
@@ -26,7 +26,7 @@ namespace Tapeti.Tasks
         {
             lock (previousTaskLock)
             {
-                previousTask = previousTask.ContinueWith(t => action(), CancellationToken.None
+                previousTask = previousTask.ContinueWith(_ => action(), CancellationToken.None
                     , TaskContinuationOptions.None
                     , singleThreadScheduler.Value);
 
@@ -43,7 +43,7 @@ namespace Tapeti.Tasks
         {
             lock (previousTaskLock)
             {
-                var task = previousTask.ContinueWith(t => func(), CancellationToken.None
+                var task = previousTask.ContinueWith(_ => func(), CancellationToken.None
                     , TaskContinuationOptions.None
                     , singleThreadScheduler.Value);
 
@@ -70,7 +70,7 @@ namespace Tapeti.Tasks
             public override int MaximumConcurrencyLevel => 1;
 
 
-            private readonly Queue<Task> scheduledTasks = new Queue<Task>();
+            private readonly Queue<Task> scheduledTasks = new();
             private bool disposed;