Suppress context of threads of ClientCommandDispatcher and ConsumerDispatcher (#1046)

* ConsumerDispatcher refactor
- add logs
- addUseWaitJoin option
- suppress flow of execution context

* remove UseWaitJoin

* remove try/catch

* tune ClientCommandDispatcherSingleton

* Info -> Debug

* add logs

* fix test

* indentation

Author: sungam3r

Source/EasyNetQ.Tests/ClientCommandDispatcherTests/When_an_action_is_invoked.cs

@@ -64,9 +64,9 @@ public void Should_invoke_the_action()
         [Fact]
         public void Should_invoke_the_action_on_the_dispatcher_thread()
         {
-            actionThreadName.Should().Be("Client Command Dispatcher Thread");
+            actionThreadName.Should().Be("EasyNetQ client command dispatch thread");
         }
     }
 }
 
-// ReSharper restore InconsistentNaming
+// ReSharper restore InconsistentNaming

Source/EasyNetQ/Consumer/ConsumerDispatcher.cs

@@ -16,39 +16,46 @@ public ConsumerDispatcher(ConnectionConfiguration configuration)
         {
             Preconditions.CheckNotNull(configuration, "configuration");
 
-            var thread = new Thread(_ =>
+            using (ExecutionContext.SuppressFlow())
             {
-                var blockingCollections = new[] {durableActions, transientActions};
-                while (!cancellation.IsCancellationRequested)
-                    try
-                    {
-                        BlockingCollection<Action>.TakeFromAny(
-                            blockingCollections, out var action, cancellation.Token
-                        );
-                        action();
-                    }
-                    catch (OperationCanceledException) when (cancellation.IsCancellationRequested)
-                    {
-                        break;
-                    }
-                    catch (Exception exception)
-                    {
-                        logger.ErrorException(string.Empty, exception);
-                    }
-
-                while (BlockingCollection<Action>.TryTakeFromAny(blockingCollections, out var action) >= 0)
+                var thread = new Thread(_ =>
                 {
-                    try
-                    {
-                        action();
-                    }
-                    catch (Exception exception)
+                    var blockingCollections = new[] { durableActions, transientActions };
+                    while (!cancellation.IsCancellationRequested)
+                        try
+                        {
+                            BlockingCollection<Action>.TakeFromAny(
+                                blockingCollections, out var action, cancellation.Token
+                            );
+                            action();
+                        }
+                        catch (OperationCanceledException) when (cancellation.IsCancellationRequested)
+                        {
+                            break;
+                        }
+                        catch (Exception exception)
+                        {
+                            logger.ErrorException(string.Empty, exception);
+                        }
+
+                    while (BlockingCollection<Action>.TryTakeFromAny(blockingCollections, out var action) >= 0)
                     {
-                        logger.ErrorException(string.Empty, exception);
+                        try
+                        {
+                            action();
+                        }
+                        catch (Exception exception)
+                        {
+                            logger.ErrorException(string.Empty, exception);
+                        }
                     }
-                }
-            }) {Name = "EasyNetQ consumer dispatch thread", IsBackground = configuration.UseBackgroundThreads};
-            thread.Start();
+                    logger.Debug("EasyNetQ consumer dispatch thread finished");
+                })
+                { Name = "EasyNetQ consumer dispatch thread", IsBackground = configuration.UseBackgroundThreads };
+
+                thread.Start();
+                logger.Debug("EasyNetQ consumer dispatch thread started");
+            }
         }
 
         public void QueueAction(Action action, bool surviveDisconnect = false)
@@ -66,11 +73,17 @@ public void QueueAction(Action action, bool surviveDisconnect = false)
 
         public void OnDisconnected()
         {
+            int count = 0;
+
             // throw away any queued actions. RabbitMQ will redeliver any in-flight
             // messages that have not been acked when the connection is lost.
             while (transientActions.TryTake(out _))
             {
+                ++count;
             }
+
+            if (count > 0)
+                logger.Debug("{count} queued transient actions were thrown", count);
         }
 
         public void Dispose()

Source/EasyNetQ/Producer/ClientCommandDispatcher.cs

@@ -48,7 +48,7 @@ public Task InvokeAsync(Action<IModel> channelAction)
 
         public void Dispose()
         {
-            if(dispatcher.IsValueCreated) dispatcher.Value.Dispose();
+            if (dispatcher.IsValueCreated) dispatcher.Value.Dispose();
         }
     }
-}
+}

Source/EasyNetQ/Producer/ClientCommandDispatcherSingleton.cs

@@ -1,14 +1,15 @@
-using System;
+using EasyNetQ.Logging;
+using RabbitMQ.Client;
+using System;
 using System.Collections.Concurrent;
 using System.Threading;
 using System.Threading.Tasks;
-using EasyNetQ.Internals;
-using RabbitMQ.Client;
 
 namespace EasyNetQ.Producer
 {
     public class ClientCommandDispatcherSingleton : IClientCommandDispatcher
     {
+        private readonly ILog logger = LogProvider.For<ClientCommandDispatcherSingleton>();
         private readonly CancellationTokenSource cancellation = new CancellationTokenSource();
         private readonly IPersistentChannel persistentChannel;
         private readonly BlockingCollection<Action> queue;
@@ -25,7 +26,8 @@ public ClientCommandDispatcherSingleton(
             queue = new BlockingCollection<Action>(configuration.DispatcherQueueSize);
             persistentChannel = persistentChannelFactory.CreatePersistentChannel(connection);
 
-            StartDispatcherThread(configuration);
+            using (ExecutionContext.SuppressFlow())
+                StartDispatcherThread(configuration);
         }
 
         public T Invoke<T>(Func<IModel, T> channelAction)
@@ -86,6 +88,7 @@ public Task InvokeAsync(Action<IModel> channelAction)
 
         public void Dispose()
         {
+            queue.CompleteAdding();
             cancellation.Cancel();
             persistentChannel.Dispose();
         }
@@ -101,13 +104,16 @@ private void StartDispatcherThread(ConnectionConfiguration configuration)
                         var channelAction = queue.Take(cancellation.Token);
                         channelAction();
                     }
-                    catch (OperationCanceledException)
+                    catch (OperationCanceledException) when (cancellation.IsCancellationRequested)
                     {
                         break;
                     }
                 }
-            }) {Name = "Client Command Dispatcher Thread", IsBackground = configuration.UseBackgroundThreads};
+                logger.Debug("EasyNetQ client command dispatch thread finished");
+            })
+            { Name = "EasyNetQ client command dispatch thread", IsBackground = configuration.UseBackgroundThreads };
             thread.Start();
+            logger.Debug("EasyNetQ client command dispatch thread started");
         }
 
         private struct NoContentStruct

hall_of_fame.md

@@ -66,4 +66,5 @@ No particular order. Don't forget to add your name with your pull request.
 * Alina Popa
 * Connie Yau
 * Marcus Hellsten
-* Thomas Mutton
+* Thomas Mutton
+* Ivan Maximov