dotnet-client observers 0.9.1 channel shutdown protocol

Author: Alexandru Scvortov

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

@@ -786,12 +786,13 @@ public void NotifyReceivedCloseOk()
             TerminateMainloop();
             m_closed = true;
         }
-        
+
         ///<summary>
         /// Sets the channel named in the SoftProtocolException into
         /// "quiescing mode", where we issue a channel.close and
-        /// ignore everything up to the channel.close-ok reply that
-        /// should eventually arrive.
+        /// ignore everything except for subsequent channel.close
+        /// messages and the channel.close-ok reply that should
+        /// eventually arrive.
         ///</summary>
         ///<remarks>
         ///<para>
@@ -816,23 +817,12 @@ public void NotifyReceivedCloseOk()
         ///</para>
         ///</remarks>
         public void QuiesceChannel(SoftProtocolException pe) {
-            // First, construct the close request and QuiescingSession
-            // that we'll use during the quiesce process.
-
-            Command request;
-            int replyClassId;
-            int replyMethodId;
-            Protocol.CreateChannelClose(pe.ReplyCode,
-                                        pe.Message,
-                                        out request,
-                                        out replyClassId,
-                                        out replyMethodId);
+            // Construct the QuiescingSession that we'll use during
+            // the quiesce process.
 
             ISession newSession = new QuiescingSession(this,
                                                        pe.Channel,
-                                                       pe.ShutdownReason,
-                                                       replyClassId,
-                                                       replyMethodId);
+                                                       pe.ShutdownReason);
 
             // Here we detach the session from the connection. It's
             // still alive: it just won't receive any further frames
@@ -851,7 +841,7 @@ public void QuiesceChannel(SoftProtocolException pe) {
             // our peer. The peer will respond through the lower
             // layers - specifically, through the QuiescingSession we
             // installed above.
-            newSession.Transmit(request);
+            newSession.Transmit(ChannelCloseWrapper(pe.ReplyCode, pe.Message));
         }
 
         public void HandleMainLoopException(ShutdownEventArgs reason) {
@@ -959,7 +949,19 @@ public Command ConnectionCloseWrapper(ushort reasonCode, string reasonText)
                                            out replyClassId,
                                            out replyMethodId);
             return request;
-        } 
+        }
+
+        protected Command ChannelCloseWrapper(ushort reasonCode, string reasonText)
+        {
+            Command request;
+            int replyClassId, replyMethodId;
+            Protocol.CreateChannelClose(reasonCode,
+                                        reasonText,
+                                        out request,
+                                        out replyClassId,
+                                        out replyMethodId);
+            return request;
+        }
 
         private static uint NegotiatedMaxValue(uint clientValue, uint serverValue)
         {

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

@@ -65,45 +65,60 @@
 // the versions we support*. Obviously we may need to revisit this if
 // that ever changes.
 using CommonFraming = RabbitMQ.Client.Framing.v0_9;
+using CommonFramingSpecs = RabbitMQ.Client.Framing.Impl.v0_9;
 
 namespace RabbitMQ.Client.Impl
 {
     ///<summary>Small ISession implementation used during channel quiescing.</summary>
     public class QuiescingSession: SessionBase
     {
         public ShutdownEventArgs m_reason;
-        public int m_replyClassId;
-        public int m_replyMethodId;
+        protected int m_closeClassId;
+        protected int m_closeMethodId;
+        protected int m_closeOkClassId;
+        protected int m_closeOkMethodId;
 
         public QuiescingSession(ConnectionBase connection,
                                 int channelNumber,
-                                ShutdownEventArgs reason,
-                                int replyClassId,
-                                int replyMethodId)
+                                ShutdownEventArgs reason)
             : base(connection, channelNumber)
         {
             m_reason = reason;
-            m_replyClassId = replyClassId;
-            m_replyMethodId = replyMethodId;
+            m_closeClassId = CommonFramingSpecs.ChannelClose.ClassId;
+            m_closeMethodId = CommonFramingSpecs.ChannelClose.MethodId;
+            m_closeOkClassId = CommonFramingSpecs.ChannelCloseOk.ClassId;
+            m_closeOkMethodId = CommonFramingSpecs.ChannelCloseOk.MethodId;
         }
 
         public override void HandleFrame(Frame frame)
         {
             if (frame.Type == CommonFraming.Constants.FrameMethod) {
                 MethodBase method = Connection.Protocol.DecodeMethodFrom(frame.GetReader());
-                if ((method.ProtocolClassId == m_replyClassId)
-                    && (method.ProtocolMethodId == m_replyMethodId))
+                if ((method.ProtocolClassId == m_closeOkClassId)
+                    && (method.ProtocolMethodId == m_closeOkMethodId))
                 {
                     // This is the reply we were looking for. Release
                     // the channel with the reason we were passed in
                     // our constructor.
                     Close(m_reason);
                     return;
                 }
+                else if ((method.ProtocolClassId == m_closeClassId)
+                         && (method.ProtocolMethodId == m_closeMethodId))
+                {
+                    // We're already shutting down the channel, so
+                    // just send back an ok.
+                    Transmit(CreateChannelCloseOk());
+                    return;
+                }
             }
 
             // Either a non-method frame, or not what we were looking
             // for. Ignore it - we're quiescing.
         }
+
+        protected Command CreateChannelCloseOk() {
+            return new Command(new CommonFramingSpecs.ConnectionCloseOk());
+        }
     }
 }