Merge branch 'stable'

Author: michaelklishin

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

@@ -809,7 +809,7 @@ public void HandleBasicGetEmpty()
             k.HandleCommand(null); // release the continuation.
         }
 
-        public void HandleBasicGetOk(ulong deliveryTag,
+        public virtual void HandleBasicGetOk(ulong deliveryTag,
             bool redelivered,
             string exchange,
             string routingKey,

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

@@ -60,6 +60,29 @@ public void InheritOffsetFrom(RecoveryAwareModel other)
             MaxSeenDeliveryTag = 0;
         }
 
+        public override void HandleBasicGetOk(ulong deliveryTag,
+            bool redelivered,
+            string exchange,
+            string routingKey,
+            uint messageCount,
+            IBasicProperties basicProperties,
+            byte[] body)
+        {
+            if (deliveryTag > MaxSeenDeliveryTag)
+            {
+                MaxSeenDeliveryTag = deliveryTag;
+            }
+
+            base.HandleBasicGetOk(
+                OffsetDeliveryTag(deliveryTag),
+                redelivered,
+                exchange,
+                routingKey,
+                messageCount,
+                basicProperties,
+                body);
+        }
+
         public override void HandleBasicDeliver(string consumerTag,
             ulong deliveryTag,
             bool redelivered,
@@ -86,7 +109,7 @@ public override void BasicAck(ulong deliveryTag,
             bool multiple)
         {
             ulong realTag = deliveryTag - ActiveDeliveryTagOffset;
-            if (realTag > 0)
+            if (realTag > 0 && realTag <= deliveryTag)
             {
                 base.BasicAck(realTag, multiple);
             }
@@ -97,7 +120,7 @@ public override void BasicNack(ulong deliveryTag,
             bool requeue)
         {
             ulong realTag = deliveryTag - ActiveDeliveryTagOffset;
-            if (realTag > 0)
+            if (realTag > 0 && realTag <= deliveryTag)
             {
                 base.BasicNack(realTag, multiple, requeue);
             }
@@ -107,7 +130,7 @@ public override void BasicReject(ulong deliveryTag,
             bool requeue)
         {
             ulong realTag = deliveryTag - ActiveDeliveryTagOffset;
-            if (realTag > 0)
+            if (realTag > 0 && realTag <= deliveryTag)
             {
                 base.BasicReject(realTag, requeue);
             }

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

@@ -88,11 +88,30 @@ public override void Init()
         public void TestBasicAckAfterChannelRecovery()
         {
             var latch = new ManualResetEvent(false);
-            var cons = new AckingBasicConsumer(Model, latch, () => { CloseAndWaitForRecovery(); });
+            var cons = new AckingBasicConsumer(Model, latch, CloseAndWaitForRecovery);
 
             TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
         }
 
+        [Test]
+        public void TestBasicAckAfterBasicGetAndChannelRecovery()
+        {
+            var q = GenerateQueueName();
+            Model.QueueDeclare(q, false, false, false, null);
+            // create an offset
+            var bp = Model.CreateBasicProperties();
+            Model.BasicPublish("", q, bp, new byte [] {});
+            Thread.Sleep(50);
+            var g = Model.BasicGet(q, false);
+            CloseAndWaitForRecovery();
+            Assert.IsTrue(Conn.IsOpen);
+            Assert.IsTrue(Model.IsOpen);
+            // ack the message after recovery - this should be out of range and ignored
+            Model.BasicAck(g.DeliveryTag, false);
+            // do a sync operation to 'check' there is no channel exception 
+            Model.BasicGet(q, false);
+        }
+
         [Test]
         public void TestBasicAckEventHandlerRecovery()
         {
@@ -200,7 +219,7 @@ public void TestBasicModelRecoveryOnServerRestart()
         public void TestBasicNackAfterChannelRecovery()
         {
             var latch = new ManualResetEvent(false);
-            var cons = new NackingBasicConsumer(Model, latch, () => { CloseAndWaitForRecovery(); });
+            var cons = new NackingBasicConsumer(Model, latch, CloseAndWaitForRecovery);
 
             TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
         }
@@ -209,7 +228,7 @@ public void TestBasicNackAfterChannelRecovery()
         public void TestBasicRejectAfterChannelRecovery()
         {
             var latch = new ManualResetEvent(false);
-            var cons = new RejectingBasicConsumer(Model, latch, () => { CloseAndWaitForRecovery(); });
+            var cons = new RejectingBasicConsumer(Model, latch, CloseAndWaitForRecovery);
 
             TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
         }