merge bug 24538 into default (WaitForConfirms with timeouts)

Author: Alexandru Scvortov

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()
+        public bool WaitForConfirms(TimeSpan timeout, out bool timedOut)
         {
-            lock (m_unconfirmedSet.SyncRoot) {
-                while (true) {
+            bool isWaitInfinite = (timeout.TotalMilliseconds == Timeout.Infinite);
+            Stopwatch stopwatch = Stopwatch.StartNew();
+            lock (m_unconfirmedSet.SyncRoot)
+            {
+                while (true)
+                {
                     if (CloseReason != null)
                         throw new AlreadyClosedException(CloseReason);
 
-                    if (m_unconfirmedSet.Count == 0) {
+                    if (m_unconfirmedSet.Count == 0)
+                    {
                         bool aux = m_onlyAcksReceived;
                         m_onlyAcksReceived = true;
+                        timedOut = false;
                         return aux;
                     }
-                    Monitor.Wait(m_unconfirmedSet.SyncRoot);
+                    if (isWaitInfinite)
+                        Monitor.Wait(m_unconfirmedSet.SyncRoot);
+                    else
+                    {
+                        TimeSpan elapsed = stopwatch.Elapsed;
+                        if(elapsed > timeout || !Monitor.Wait(
+                            m_unconfirmedSet.SyncRoot, timeout - elapsed))
+                        {
+                            timedOut = true;
+                            return true;
+                        }
+                    }
                 }
             }
         }
 
+        public bool WaitForConfirms()
+        {
+            bool timedOut;
+            return WaitForConfirms(TimeSpan.FromMilliseconds(Timeout.Infinite), out timedOut);
+        }
+
         public void WaitForConfirmsOrDie()
         {
-            if (!WaitForConfirms()) {
+            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);