Dispose consumer cancellation for RPC response queue (#969)

* fixes #968

* relates to #967

Author: sungam3r

Source/EasyNetQ/Producer/Rpc.cs

@@ -23,7 +23,7 @@ public class Rpc : IRpc
         private readonly ITypeNameSerializer typeNameSerializer;
 
         private readonly object responseQueuesAddLock = new object();
-        private readonly ConcurrentDictionary<RpcKey, string> responseQueues = new ConcurrentDictionary<RpcKey, string>();
+        private readonly ConcurrentDictionary<RpcKey, ResponseQueueWithCancellation> responseQueues = new ConcurrentDictionary<RpcKey, ResponseQueueWithCancellation>();
         private readonly ConcurrentDictionary<string, ResponseAction> responseActions = new ConcurrentDictionary<string, ResponseAction>();
 
         protected readonly TimeSpan disablePeriodicSignaling = TimeSpan.FromMilliseconds(-1);
@@ -63,11 +63,19 @@ public Rpc(
 
         private void OnConnectionCreated(ConnectionCreatedEvent @event)
         {
+            var copyOfResponseQueues = responseQueues.Values;
+            responseQueues.Clear();
+
             var copyOfResponseActions = responseActions.Values;
             responseActions.Clear();
-            responseQueues.Clear();
 
-            // retry in-flight requests.
+            // call consumer cancellations
+            foreach (var queueWithCancellation in copyOfResponseQueues)
+            {
+                queueWithCancellation.Cancellation.Dispose();
+            }
+
+            // finish in-flight requests
             foreach (var responseAction in copyOfResponseActions)
             {
                 responseAction.OnFailure();
@@ -85,7 +93,11 @@ public virtual Task<TResponse> Request<TRequest, TResponse>(TRequest request, Ac
             var configuration = new RequestConfiguration();
             configure(configuration);
 
+#if NETFX && !NET46
             var tcs = new TaskCompletionSource<TResponse>();
+#else
+            var tcs = new TaskCompletionSource<TResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
+#endif
 
             var timeout = timeoutStrategy.GetTimeoutSeconds(requestType);
             Timer timer = null;
@@ -154,13 +166,12 @@ protected virtual string SubscribeToResponse<TRequest, TResponse>()
         {
             var responseType = typeof(TResponse);
             var rpcKey = new RpcKey { Request = typeof(TRequest), Response = responseType };
-            string queueName;
-            if (responseQueues.TryGetValue(rpcKey, out queueName))
-                return queueName;
+            if (responseQueues.TryGetValue(rpcKey, out ResponseQueueWithCancellation queueWithCancellation))
+                return queueWithCancellation.QueueName;
             lock (responseQueuesAddLock)
             {
-                if (responseQueues.TryGetValue(rpcKey, out queueName))
-                    return queueName;
+                if (responseQueues.TryGetValue(rpcKey, out queueWithCancellation))
+                    return queueWithCancellation.QueueName;
 
                 var queue = advancedBus.QueueDeclare(
                             conventions.RpcReturnQueueNamingConvention(),
@@ -174,15 +185,15 @@ protected virtual string SubscribeToResponse<TRequest, TResponse>()
                     queue,
                     queue.Name);
 
-                advancedBus.Consume<TResponse>(queue, (message, messageReceivedInfo) => Task.Factory.StartNew(() =>
+                var cancellation = advancedBus.Consume<TResponse>(queue, (message, messageReceivedInfo) => Task.Factory.StartNew(() =>
                     {
                         ResponseAction responseAction;
                         if (responseActions.TryRemove(message.Properties.CorrelationId, out responseAction))
                         {
                             responseAction.OnSuccess(message);
                         }
                     }));
-                responseQueues.TryAdd(rpcKey, queue.Name);
+                responseQueues.TryAdd(rpcKey, new ResponseQueueWithCancellation { QueueName = queue.Name, Cancellation = cancellation });
                 return queue.Name;
             }
         }
@@ -193,6 +204,12 @@ protected struct RpcKey
             public Type Response;
         }
 
+        protected struct ResponseQueueWithCancellation
+        {
+            public string QueueName;
+            public IDisposable Cancellation;
+        }
+
         protected class ResponseAction
         {
             public Action<object> OnSuccess { get; set; }
@@ -257,7 +274,11 @@ protected Task ExecuteResponder<TRequest, TResponse>(Func<TRequest, Task<TRespon
             where TRequest : class
             where TResponse : class
         {
+#if NETFX && !NET46
             var tcs = new TaskCompletionSource<object>();
+#else
+            var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
+#endif
 
             try
             {
@@ -344,4 +365,4 @@ private IExchange DeclareAndBindRpcExchange(string exchangeName, IQueue queue, s
             return exchange;
         }
     }
-}
+}