Introduce continuation timeouts

Author: michaelklishin

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

@@ -172,6 +172,29 @@ public class ConnectionFactory : ConnectionFactoryBase, IConnectionFactory
         /// </summary>
         public TimeSpan NetworkRecoveryInterval = TimeSpan.FromSeconds(5);
 
+        private TimeSpan m_handshakeContinuationTimeout = TimeSpan.FromSeconds(10);
+        private TimeSpan m_continuationTimeout = TimeSpan.FromSeconds(20);
+
+        /// <summary>
+        /// Amount of time protocol handshake operations are allowed to take before
+        /// timing out.
+        /// </summary>
+        public TimeSpan HandshakeContinuationTimeout
+        {
+            get { return m_handshakeContinuationTimeout; }
+            set { m_handshakeContinuationTimeout = value; }
+        }
+
+        /// <summary>
+        /// Amount of time protocol  operations (e.g. <code>queue.declare</code>) are allowed to take before
+        /// timing out.
+        /// </summary>
+        public TimeSpan ContinuationTimeout
+        {
+            get { return m_continuationTimeout; }
+            set { m_continuationTimeout = value; }
+        }
+
         /// <summary>
         /// The port to connect on. <see cref="AmqpTcpEndpoint.UseDefaultPort"/>
         ///  indicates the default for the protocol should be used.

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

@@ -110,5 +110,17 @@ public interface IConnectionFactory
         /// What task scheduler should consumer dispatcher use.
         /// </summary>
         TaskScheduler TaskScheduler { get; set; }
+
+        /// <summary>
+        /// Amount of time protocol handshake operations are allowed to take before
+        /// timing out.
+        /// </summary>
+        TimeSpan HandshakeContinuationTimeout { get; set; }
+
+        /// <summary>
+        /// Amount of time protocol  operations (e.g. <code>queue.declare</code>) are allowed to take before
+        /// timing out.
+        /// </summary>
+        TimeSpan ContinuationTimeout { get; set; }
     }
 }

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

@@ -611,5 +611,11 @@ void QueueDeclareNoWait(string queue, bool durable, bool exclusive,
         /// </remarks>
         [AmqpMethodDoNotImplement(null)]
         void WaitForConfirmsOrDie(TimeSpan timeout);
+
+        /// <summary>
+        /// Amount of time protocol  operations (e.g. <code>queue.declare</code>) are allowed to take before
+        /// timing out.
+        /// </summary>
+        TimeSpan ContinuationTimeout { get; set; }
     }
 }

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

@@ -77,6 +77,12 @@ public IConsumerDispatcher ConsumerDispatcher
             get { return m_delegate.ConsumerDispatcher; }
         }
 
+        public TimeSpan ContinuationTimeout
+        {
+            get { return m_delegate.ContinuationTimeout; }
+            set { m_delegate.ContinuationTimeout = value; }
+        }
+
         public AutorecoveringModel(AutorecoveringConnection conn, RecoveryAwareModel _delegate)
         {
             m_connection = conn;

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

@@ -1204,6 +1204,7 @@ public IModel CreateModel()
             EnsureIsOpen();
             ISession session = CreateSession();
             var model = (IFullModel)Protocol.CreateModel(session, this.ConsumerWorkService);
+            model.ContinuationTimeout = m_factory.ContinuationTimeout;
             model._Private_ChannelOpen("");
             return model;
         }

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

@@ -63,6 +63,9 @@ public abstract class ModelBase : IFullModel, IRecoverable
         ///sequence. See <see cref="Connection.Open"/> </summary>
         public BlockingCell m_connectionStartCell = null;
 
+        private TimeSpan m_handshakeContinuationTimeout = TimeSpan.FromSeconds(10);
+        private TimeSpan m_continuationTimeout = TimeSpan.FromSeconds(20);
+
         public RpcContinuationQueue m_continuationQueue = new RpcContinuationQueue();
         public ManualResetEvent m_flowControlBlock = new ManualResetEvent(true);
 
@@ -106,6 +109,18 @@ protected void Initialise(ISession session)
             Session.SessionShutdown += OnSessionShutdown;
         }
 
+        public TimeSpan HandshakeContinuationTimeout
+        {
+            get { return m_handshakeContinuationTimeout; }
+            set { m_handshakeContinuationTimeout = value; }
+        }
+
+        public TimeSpan ContinuationTimeout
+        {
+            get { return m_continuationTimeout; }
+            set { m_continuationTimeout = value; }
+        }
+
         public event EventHandler<BasicAckEventArgs> BasicAcks
         {
             add
@@ -300,8 +315,8 @@ public void Close(ShutdownEventArgs reason, bool abort)
                 if (SetCloseReason(reason))
                 {
                     _Private_ChannelClose(reason.ReplyCode, reason.ReplyText, 0, 0);
-                }
-                k.Wait();
+                }                
+                k.Wait(TimeSpan.FromMilliseconds(10000));
                 ConsumerDispatcher.Shutdown(this);
             }
             catch (AlreadyClosedException)
@@ -343,7 +358,7 @@ public string ConnectionOpen(string virtualHost,
                 // which is a much more suitable exception before connection
                 // negotiation finishes
             }
-            k.GetReply();
+            k.GetReply(HandshakeContinuationTimeout);
             return k.m_knownHosts;
         }
 
@@ -361,7 +376,7 @@ public ConnectionSecureOrTune ConnectionSecureOk(byte[] response)
                 // which is a much more suitable exception before connection
                 // negotiation finishes
             }
-            k.GetReply();
+            k.GetReply(HandshakeContinuationTimeout);
             return k.m_result;
         }
 
@@ -383,7 +398,7 @@ public ConnectionSecureOrTune ConnectionStartOk(IDictionary<string, object> clie
                 // which is a much more suitable exception before connection
                 // negotiation finishes
             }
-            k.GetReply();
+            k.GetReply(HandshakeContinuationTimeout);
             return k.m_result;
         }
 
@@ -434,7 +449,7 @@ public MethodBase ModelRpc(MethodBase method, ContentHeaderBase header, byte[] b
         {
             var k = new SimpleBlockingRpcContinuation();
             TransmitAndEnqueue(new Command(method, header, body), k);
-            return k.GetReply().Method;
+            return k.GetReply(this.ContinuationTimeout).Method;
         }
 
         public void ModelSend(MethodBase method, ContentHeaderBase header, byte[] body)
@@ -1127,7 +1142,7 @@ public void BasicCancel(string consumerTag)
             Enqueue(k);
 
             _Private_BasicCancel(consumerTag, false);
-            k.GetReply();
+            k.GetReply(this.ContinuationTimeout);
             lock (m_consumers)
             {
                 m_consumers.Remove(consumerTag);
@@ -1175,7 +1190,7 @@ public string BasicConsume(string queue,
             // the RPC response, but a response is still expected.
             _Private_BasicConsume(queue, consumerTag, noLocal, noAck, exclusive,
                 /*nowait:*/ false, arguments);
-            k.GetReply();
+            k.GetReply(this.ContinuationTimeout);
             string actualConsumerTag = k.m_consumerTag;
 
             return actualConsumerTag;
@@ -1187,7 +1202,7 @@ public BasicGetResult BasicGet(string queue,
             var k = new BasicGetRpcContinuation();
             Enqueue(k);
             _Private_BasicGet(queue, noAck);
-            k.GetReply();
+            k.GetReply(this.ContinuationTimeout);
             return k.m_result;
         }
 
@@ -1255,7 +1270,7 @@ public void BasicRecover(bool requeue)
 
             Enqueue(k);
             _Private_BasicRecover(requeue);
-            k.GetReply();
+            k.GetReply(this.ContinuationTimeout);
         }
 
         public abstract void BasicRecoverAsync(bool requeue);
@@ -1586,7 +1601,7 @@ private QueueDeclareOk QueueDeclare(string queue, bool passive, bool durable, bo
             var k = new QueueDeclareRpcContinuation();
             Enqueue(k);
             _Private_QueueDeclare(queue, passive, durable, exclusive, autoDelete, false, arguments);
-            k.GetReply();
+            k.GetReply(this.ContinuationTimeout);
             return k.m_result;
         }
 

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

@@ -76,5 +76,10 @@ public virtual ShutdownEventArgs Wait()
         {
             return (ShutdownEventArgs)m_cell.Value;
         }
+
+        public ShutdownEventArgs Wait(TimeSpan timeout)
+        {
+            return (ShutdownEventArgs)m_cell.GetValue(timeout);
+        }
     }
 }

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

@@ -69,6 +69,31 @@ public virtual Command GetReply()
             }
         }
 
+        public virtual Command GetReply(TimeSpan timeout)
+        {
+            var result = (Either)m_cell.GetValue(timeout);
+            switch (result.Alternative)
+            {
+                case EitherAlternative.Left:
+                    return (Command)result.Value;
+                case EitherAlternative.Right:
+                    throw new OperationInterruptedException((ShutdownEventArgs)result.Value);
+                default:
+                    ReportInvalidInvariant(result);
+                    return null;
+            }
+        }
+
+        private static void ReportInvalidInvariant(Either result)
+        {
+            string error = "Illegal EitherAlternative " + result.Alternative;
+#if !(NETFX_CORE)
+            Trace.Fail(error);
+#else
+            MetroEventSource.Log.Error(error);
+#endif
+        }
+
         public virtual void HandleCommand(Command cmd)
         {
             m_cell.Value = Either.Left(cmd);