Merging bug22814 (race in handling (synchronous) basic.recover, .net client)

Author: Simon MacMullen

projects/client/ApigenBootstrap/RabbitMQ.Client.ApigenBootstrap.csproj

@@ -72,6 +72,9 @@
     <Compile Include="..\RabbitMQ.Client\src\client\api\ShutdownInitiator.cs">
       <Link>src\client\api\ShutdownInitiator.cs</Link>
     </Compile>
+    <Compile Include="..\RabbitMQ.Client\src\client\events\BasicRecoverOkEventHandler.cs">
+      <Link>src\client\events\BasicRecoverOkEventHandler.cs</Link>
+    </Compile>
     <Compile Include="..\RabbitMQ.Client\src\client\events\BasicReturnEventArgs.cs">
       <Link>src\client\events\BasicReturnEventArgs.cs</Link>
     </Compile>

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

@@ -93,6 +93,16 @@ public interface IModel: IDisposable
         ///</remarks>
         event CallbackExceptionEventHandler CallbackException;
 
+        ///<summary>All messages received before this fires that haven't been
+        ///ack'ed will be redelivered. All messages received afterwards won't
+        ///be.
+        ///
+        ///Handlers for this event are invoked by the connection thread.
+        ///It is sometimes useful to allow that thread to know that a recover-ok
+        ///has been received, rather than the thread that invoked BasicRecover().
+        ///</summary>
+        event BasicRecoverOkEventHandler BasicRecoverOk;
+
         ///<summary>Signalled when an unexpected message is delivered
         ///
         /// Under certain circumstances it is possible for a channel to receive a
@@ -364,6 +374,7 @@ void BasicReject(ulong deliveryTag,
                          bool requeue);
 
         ///<summary>(Spec method)</summary>
+        [AmqpMethodDoNotImplement(null)]
         void BasicRecover(bool requeue);
 
         ///<summary>(Spec method)</summary>
@@ -622,6 +633,15 @@ void HandleBasicGetOk(ulong deliveryTag,
         ///</remarks>
         void HandleBasicGetEmpty();
 
+        ///<summary>Handle incoming Basic.RecoverOk methods
+        ///received in reply to Basic.Recover.
+        ///</summary>
+        void HandleBasicRecoverOk();
+
+        [AmqpForceOneWay]
+        [AmqpMethodMapping(null, "basic", "recover")]
+        void _Private_BasicRecover(bool requeue);
+
         ///<summary>Handle incoming Basic.Deliver methods. Dispatches
         ///to waiting consumers.</summary>
         void HandleBasicDeliver(string consumerTag,

projects/client/RabbitMQ.Client/src/client/events/BasicRecoverOkEventHandler.cs

@@ -0,0 +1,64 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 1.1.
+//
+// The APL v2.0:
+//
+//---------------------------------------------------------------------------
+//   Copyright (C) 2007-2010 LShift Ltd., Cohesive Financial
+//   Technologies LLC., and Rabbit Technologies Ltd.
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//---------------------------------------------------------------------------
+//
+// The MPL v1.1:
+//
+//---------------------------------------------------------------------------
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.rabbitmq.com/mpl.html
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is The RabbitMQ .NET Client.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+//---------------------------------------------------------------------------
+using System;
+
+namespace RabbitMQ.Client.Events
+{
+    ///<summary>Delegate used to process Basic.RecoverOk events.</summary>
+    public delegate void BasicRecoverOkEventHandler(IModel model, EventArgs args);
+}
+

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

@@ -82,6 +82,7 @@ public abstract class ModelBase : IFullModel
         private readonly object m_eventLock = new object();
         private BasicReturnEventHandler m_basicReturn;
         private CallbackExceptionEventHandler m_callbackException;
+        private BasicRecoverOkEventHandler m_basicRecoverOk;
 
         public ManualResetEvent m_flowControlBlock = new ManualResetEvent(true);
 
@@ -148,6 +149,24 @@ public event CallbackExceptionEventHandler CallbackException
             }
         }
 
+        public event BasicRecoverOkEventHandler BasicRecoverOk
+        {
+            add
+            {
+                lock (m_eventLock)
+                {
+                    m_basicRecoverOk += value;
+                }
+            }
+            remove
+            {
+                lock (m_eventLock)
+                {
+                    m_basicRecoverOk -= value;
+                }
+            }
+        }
+
         public IBasicConsumer DefaultConsumer { get; set; }
 
         public ISession m_session;
@@ -278,6 +297,31 @@ public virtual void OnCallbackException(CallbackExceptionEventArgs args)
                 }
             }
         }
+
+        public virtual void OnBasicRecoverOk(EventArgs args)
+        {
+            BasicRecoverOkEventHandler handler;
+            lock (m_eventLock)
+            {
+                handler = m_basicRecoverOk;
+            }
+            if (handler != null)
+            {
+                foreach (BasicRecoverOkEventHandler h in handler.GetInvocationList())
+                {
+                    try
+                    {
+                        h(this, args);
+                    }
+                    catch (Exception e)
+                    {
+                        CallbackExceptionEventArgs exnArgs = new CallbackExceptionEventArgs(e);
+                        exnArgs.Detail["context"] = "OnBasicRecoverOk";
+                        OnCallbackException(exnArgs);
+                    }
+                }
+            }
+        }
 
         public void Enqueue(IRpcContinuation k)
         {
@@ -734,6 +778,28 @@ public BasicGetResult BasicGet(string queue,
             return k.m_result;
         }
 
+        public abstract void _Private_BasicRecover(bool requeue);
+
+        public void BasicRecover(bool requeue)
+        {
+            SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
+
+            Enqueue(k);
+
+            try
+            {
+                _Private_BasicRecover(requeue);
+            }
+            catch (AlreadyClosedException)
+            {
+                // Ignored, since the continuation will be told about
+                // the closure via an OperationInterruptedException because
+                // of the shutdown event propagation.
+            }
+
+            k.GetReply();
+        }
+
         public abstract void BasicQos(uint prefetchSize,
                                       ushort prefetchCount,
                                       bool global);
@@ -804,7 +870,6 @@ public abstract void BasicAck(ulong deliveryTag,
         public abstract void BasicReject(ulong deliveryTag,
                                          bool requeue);
 
-        public abstract void BasicRecover(bool requeue);
         public abstract void BasicRecoverAsync(bool requeue);
 
         public abstract void TxSelect();
@@ -904,6 +969,13 @@ public void HandleBasicGetEmpty()
             k.HandleCommand(null); // release the continuation.
         }
 
+        public void HandleBasicRecoverOk()
+        {
+            SimpleBlockingRpcContinuation k = (SimpleBlockingRpcContinuation)m_continuationQueue.Next();
+            OnBasicRecoverOk(new EventArgs());
+            k.HandleCommand(null);
+        }
+
         public abstract ConnectionTuneDetails ConnectionStartOk(IDictionary clientProperties,
                                                                 string mechanism,
                                                                 byte[] response,

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

@@ -115,5 +115,15 @@ public void TestRecoverAfterCancel_()
             Assert.IsFalse(Event.Redelivered);
             Assert.IsTrue(Event2.Redelivered);
         }
+
+        [Test]
+        public void TestRecoverCallback()
+        {
+            int callbackCount = 0;
+            Channel.BasicRecoverOk += (sender, eventArgs) => callbackCount++;
+            Channel.BasicRecover(false);
+            Assert.AreEqual(1, callbackCount);
+        }
+
     }
 }