add basic confirm barrier

Author: Alexandru Scvortov

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

@@ -317,6 +317,21 @@ uint QueueDelete(string queue,
         [AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
         void ConfirmSelect();
 
+        //<summary>Wait until all published messages have been confirmed.
+        //<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();
+
+
         ///<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

@@ -76,6 +76,9 @@ public abstract class ModelBase : IFullModel
         private readonly object m_flowSendLock = new object();
 
         private ulong m_nextPubSeqNo;
+        private SortedList m_unconfirmedSet =
+            SortedList.Synchronized(new SortedList());
+        private bool m_onlyAcksReceived = true;
 
         public event ModelShutdownEventHandler ModelShutdown
         {
@@ -352,6 +355,8 @@ public virtual void OnBasicAck(BasicAckEventArgs args)
                     }
                 }
             }
+
+            handleAckNack(args.DeliveryTag, args.Multiple, false);
         }
 
         public virtual void OnBasicNack(BasicNackEventArgs args)
@@ -373,6 +378,24 @@ public virtual void OnBasicNack(BasicNackEventArgs args)
                     }
                 }
             }
+
+            handleAckNack(args.DeliveryTag, args.Multiple, true);
+        }
+
+        protected virtual void handleAckNack(ulong deliveryTag, bool multiple, bool isNack)
+        {
+            if (multiple) {
+                for (ulong i = (ulong)m_unconfirmedSet.GetKey(0); i <= deliveryTag; i++) {
+                    m_unconfirmedSet.Remove(i);
+                }
+            } else {
+                m_unconfirmedSet.Remove(deliveryTag);
+            }
+            lock (m_unconfirmedSet.SyncRoot) {
+                m_onlyAcksReceived = m_onlyAcksReceived && !isNack;
+                if (m_unconfirmedSet.Count == 0)
+                    Monitor.Pulse(m_unconfirmedSet.SyncRoot);
+            }
         }
 
         public virtual void OnCallbackException(CallbackExceptionEventArgs args)
@@ -902,11 +925,26 @@ public abstract uint _Private_QueueDelete(string queue,
                                                   bool ifEmpty,
                                                   bool nowait);
 
-        public void ConfirmSelect() {
+        public void ConfirmSelect()
+        {
             m_nextPubSeqNo = 1;
             _Private_ConfirmSelect(false);
         }
 
+        public bool WaitForConfirms()
+        {
+            lock (m_unconfirmedSet.SyncRoot) {
+                while (true) {
+                    if (m_unconfirmedSet.Count == 0) {
+                        bool aux = m_onlyAcksReceived;
+                        m_onlyAcksReceived = true;
+                        return aux;
+                    }
+                    Monitor.Wait(m_unconfirmedSet.SyncRoot);
+                }
+            }
+        }
+
         public abstract void _Private_ConfirmSelect(bool nowait);
 
         public string BasicConsume(string queue,
@@ -1143,7 +1181,10 @@ public void BasicPublish(string exchange,
             {
                 basicProperties = CreateBasicProperties();
             }
-            if (m_nextPubSeqNo > 0) m_nextPubSeqNo++;
+            if (m_nextPubSeqNo > 0) {
+                m_unconfirmedSet.Add(m_nextPubSeqNo, null);
+                m_nextPubSeqNo++;
+            }
             _Private_BasicPublish(exchange,
                                   routingKey,
                                   mandatory,

projects/client/Unit/src/unit/TestExtensions.cs

@@ -73,5 +73,15 @@ public void TestExchangeBinding()
             Model.ExchangeDelete("src");
             Model.ExchangeDelete("dest");
         }
+
+        [Test]
+        public void TestConfirmBarrier()
+        {
+            Assert.That(Model.WaitForConfirms(), Is.Not.False);
+            Model.ConfirmSelect();
+            for (int i = 0; i < 10; i++)
+                Model.BasicPublish("", String.Empty, null, new byte[] { });
+            Assert.That(Model.WaitForConfirms(), Is.True);
+        }
     }
 }