Stop/resume flow of methods that have some content,

after we get Channel.Flow from the broker.

Author: Hubert Plociniczak

src/client/api/IModel.cs

@@ -620,6 +620,15 @@ void HandleBasicReturn(ushort replyCode,
                                IBasicProperties basicProperties,
                                [AmqpContentBodyMapping]
                                byte[] body);
+                               
+        ///<summary>Used to send a Channel.FlowOk. Confirms that
+        ///Channel.Flow from the broker was processed.</summary>
+        [AmqpMethodMapping(null, "channel", "flow-ok")]
+        void _Private_ChannelFlowOk();
+        
+        ///<summary>Handle incoming Channel.Flow methods. Either
+        ///stops or resumes sending the methods that have content.</summary>
+        void HandleChannelFlow(bool active);
 
         ///<summary>Handle an incoming Channel.Close. Shuts down the
         ///session and model.</summary>

src/client/impl/ModelBase.cs

@@ -76,6 +76,8 @@ public abstract class ModelBase : IFullModel
         private readonly object m_eventLock = new object();
         private BasicReturnEventHandler m_basicReturn;
         private CallbackExceptionEventHandler m_callbackException;
+        
+        public ManualResetEvent m_flowControlBlock = new ManualResetEvent(true);
 
         public event ModelShutdownEventHandler ModelShutdown
         {
@@ -225,6 +227,7 @@ public virtual void OnModelShutdown(ShutdownEventArgs reason)
                     }
                 }
             }
+            m_flowControlBlock.Set();
         }
 
         public virtual void OnBasicReturn(BasicReturnEventArgs args)
@@ -318,6 +321,9 @@ public bool IsOpen
 
         public void ModelSend(MethodBase method, ContentHeaderBase header, byte[] body)
         {
+            if (method.HasContent) {
+                m_flowControlBlock.WaitOne();
+            }
             m_session.Transmit(new Command(method, header, body));
         }
 
@@ -381,6 +387,17 @@ public void HandleBasicReturn(ushort replyCode,
             e.Body = body;
             OnBasicReturn(e);
         }
+        
+        public abstract void _Private_ChannelFlowOk();
+        
+        public void HandleChannelFlow(bool active)
+        {
+            if (active)
+                m_flowControlBlock.Set();
+            else
+                m_flowControlBlock.Reset();
+            _Private_ChannelFlowOk();
+        }
 
         public void HandleConnectionStart(byte versionMajor,
                                           byte versionMinor,