add WaitForConfirmsOrDie

Author: Alexandru Scvortov

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

@@ -331,6 +331,17 @@ uint QueueDelete(string queue,
         [AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
         bool WaitForConfirms();
 
+        //<summary>Wait until all published messages have been confirmed.
+        //<remarks>
+        //Waits until all messages published since the last call have
+        //been ack'd by the broker.  If a nack is received, 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();
 
         ///<summary>Start a Basic content-class consumer.</summary>
         ///<remarks>

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

@@ -312,6 +312,8 @@ public virtual void OnModelShutdown(ShutdownEventArgs reason)
                     }
                 }
             }
+            lock (m_unconfirmedSet.SyncRoot)
+                Monitor.Pulse(m_unconfirmedSet.SyncRoot);
             m_flowControlBlock.Set();
         }
 
@@ -935,6 +937,9 @@ public bool WaitForConfirms()
         {
             lock (m_unconfirmedSet.SyncRoot) {
                 while (true) {
+                    if (CloseReason != null)
+                        throw new OperationInterruptedException(CloseReason);
+
                     if (m_unconfirmedSet.Count == 0) {
                         bool aux = m_onlyAcksReceived;
                         m_onlyAcksReceived = true;
@@ -945,6 +950,17 @@ public bool WaitForConfirms()
             }
         }
 
+        public void WaitForConfirmsOrDie()
+        {
+            if (!WaitForConfirms()) {
+                Close(new ShutdownEventArgs(ShutdownInitiator.Application,
+                                            CommonFraming.Constants.ReplySuccess,
+                                            "Goodbye", new IOException("nack received")),
+                      false);
+                throw new OperationInterruptedException(CloseReason);
+            }
+        }
+
         public abstract void _Private_ConfirmSelect(bool nowait);
 
         public string BasicConsume(string queue,
@@ -1229,7 +1245,7 @@ public void Close()
         	Close(CommonFraming.Constants.ReplySuccess, "Goodbye");
         }
 
-		public void Close(ushort replyCode, string replyText)
+        public void Close(ushort replyCode, string replyText)
         {
         	Close(replyCode, replyText, false);
         }
@@ -1245,16 +1261,21 @@ public void Abort(ushort replyCode, string replyText)
         }
 
         public void Close(ushort replyCode, string replyText, bool abort)
+        {
+            Close(new ShutdownEventArgs(ShutdownInitiator.Application,
+                                        replyCode, replyText),
+                  abort);
+        }
+
+        public void Close(ShutdownEventArgs reason, bool abort)
         {
             ShutdownContinuation k = new ShutdownContinuation();
             ModelShutdown += new ModelShutdownEventHandler(k.OnShutdown);
 
             try {
-                if (SetCloseReason(new ShutdownEventArgs(ShutdownInitiator.Application,
-                                     replyCode,
-                                     replyText)))
+                if (SetCloseReason(reason))
                 {
-                    _Private_ChannelClose(replyCode, replyText, 0, 0);
+                    _Private_ChannelClose(reason.ReplyCode, reason.ReplyText, 0, 0);
                 }
                 k.Wait();
             } catch (AlreadyClosedException ace) {