Added versions of WaitForConfirms and WaitForConfirmsOrDie with timeouts

Author: jkff

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

@@ -332,6 +332,28 @@ uint QueueDelete(string queue,
         [AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
         bool WaitForConfirms();
 
+        ///<summary>Wait until all published messages have been confirmed.
+        ///</summary>
+        ///<returns>true if no nacks were received within the timeout, 
+        ///otherwise false</returns>
+        ///<param name="timeout">How long to wait (at most) before returning 
+        ///whether or not any nacks were returned</param>
+        ///<param name="timedOut">True if the method returned because
+        ///the timeout elapsed, not because all messages were ack'd
+        ///or at least one nack'd.</param>
+        ///<remarks>
+        ///Waits until all messages published since the last call have
+        ///been either ack'd or nack'd by the broker.  Returns whether
+        ///all the messages were ack'd (and none were nack'd). Note,
+        ///when called on a non-Confirm channel, returns true
+        ///immediately.
+        ///</remarks>
+        [AmqpMethodDoNotImplement(null)]
+        [AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
+        [AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
+        [AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
+        bool WaitForConfirms(TimeSpan timeout, out bool timedOut);
+
         ///<summary>Wait until all published messages have been confirmed.
         ///</summary>
         ///<remarks>
@@ -345,6 +367,20 @@ uint QueueDelete(string queue,
         [AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
         void WaitForConfirmsOrDie();
 
+        ///<summary>Wait until all published messages have been confirmed.
+        ///</summary>
+        ///<remarks>
+        ///Waits until all messages published since the last call have
+        ///been ack'd by the broker.  If a nack is received or the timeout 
+        ///elapses, throws an OperationInterrupedException exception 
+        ///immediately.
+        ///</remarks>
+        [AmqpMethodDoNotImplement(null)]
+        [AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
+        [AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
+        [AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
+        void WaitForConfirmsOrDie(TimeSpan timeout);
+
         ///<summary>Start a Basic content-class consumer.</summary>
         ///<remarks>
         ///The consumer is started with noAck=false (i.e. BasicAck is required),

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

@@ -958,32 +958,70 @@ public void ConfirmSelect()
             _Private_ConfirmSelect(false);
         }
 
-        public bool WaitForConfirms()
-        {
-            lock (m_unconfirmedSet.SyncRoot) {
-                while (true) {
-                    if (CloseReason != null)
-                        throw new AlreadyClosedException(CloseReason);
-
-                    if (m_unconfirmedSet.Count == 0) {
-                        bool aux = m_onlyAcksReceived;
-                        m_onlyAcksReceived = true;
-                        return aux;
-                    }
-                    Monitor.Wait(m_unconfirmedSet.SyncRoot);
-                }
-            }
+        public bool WaitForConfirms(TimeSpan timeout, out bool timedOut)
+        {
+            var isWaitInfinite = (timeout.TotalMilliseconds == Timeout.Infinite);
+            var stopwatch = Stopwatch.StartNew();
+            lock (m_unconfirmedSet.SyncRoot)
+            {
+                while (true)
+                {
+                    if (CloseReason != null)
+                        throw new AlreadyClosedException(CloseReason);
+
+                    if (m_unconfirmedSet.Count == 0)
+                    {
+                        bool aux = m_onlyAcksReceived;
+                        m_onlyAcksReceived = true;
+                        timedOut = false;
+                        return aux;
+                    }
+                    if (isWaitInfinite)
+                        Monitor.Wait(m_unconfirmedSet.SyncRoot);
+                    else
+                    {
+                        var elapsed = stopwatch.Elapsed;
+                        if(elapsed > timeout || !Monitor.Wait(
+                            m_unconfirmedSet.SyncRoot, timeout - elapsed))
+                        {
+                            timedOut = true;
+                            return true;
+                        }
+                    }
+                }
+            }
         }
 
-        public void WaitForConfirmsOrDie()
-        {
-            if (!WaitForConfirms()) {
+        public bool WaitForConfirms()
+        {
+            bool timedOut;
+            return WaitForConfirms(TimeSpan.FromMilliseconds(Timeout.Infinite), out timedOut);
+        }
+
+        public void WaitForConfirmsOrDie()
+        {
+            WaitForConfirmsOrDie(TimeSpan.FromMilliseconds(Timeout.Infinite));
+        }
+
+        public void WaitForConfirmsOrDie(TimeSpan timeout)
+        {
+            bool timedOut;
+            bool onlyAcksReceived = WaitForConfirms(timeout, out timedOut);
+            if (!onlyAcksReceived) {
                 Close(new ShutdownEventArgs(ShutdownInitiator.Application,
                                             CommonFraming.Constants.ReplySuccess,
                                             "Nacks Received", new IOException("nack received")),
                       false);
                 throw new IOException("Nacks Received");
             }
+            if (timedOut) {
+                Close(new ShutdownEventArgs(ShutdownInitiator.Application,
+                                            CommonFraming.Constants.ReplySuccess,
+                                            "Timed out waiting for acks", 
+                                            new IOException("timed out waiting for acks")),
+                      false);
+                throw new IOException("Timed out waiting for acks");
+            }
         }
 
         public abstract void _Private_ConfirmSelect(bool nowait);