merge bug22423 into default

Author: Matthias Radestock

src/com/rabbitmq/client/Channel.java

@@ -37,6 +37,7 @@
 import com.rabbitmq.client.AMQP.Exchange;
 import com.rabbitmq.client.AMQP.Queue;
 import com.rabbitmq.client.AMQP.Tx;
+import com.rabbitmq.client.impl.AMQImpl.Channel.FlowOk;
 
 /**
  * Public API: Interface to an AMQ channel. See the <a href="http://www.amqp.org/">spec</a> for details.
@@ -90,6 +91,14 @@ public interface Channel extends ShutdownNotifier {
      * @throws java.io.IOException if an error is encountered
      */
     void close(int closeCode, String closeMessage) throws IOException;
+    
+    /**
+     * Set flow on the channel
+     * 
+     * @param active if true, the server is asked to start sending. If false, the server is asked to stop sending.
+     * @throws IOException
+     */
+    FlowOk flow(boolean active) throws IOException;
 
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code

src/com/rabbitmq/client/impl/ChannelN.java

@@ -695,4 +695,9 @@ public Tx.RollbackOk txRollback()
     {
         return (Tx.RollbackOk) exnWrappingRpc(new Tx.Rollback()).getMethod();
     }
+
+    /** Public API - {@inheritDoc} */
+    public Channel.FlowOk flow(final boolean a) throws IOException {
+        return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow() {{active = a;}}).getMethod();        
+    }
 }

test/src/com/rabbitmq/client/test/functional/QosTests.java

@@ -359,6 +359,19 @@ public void testLimitInheritsUnackedCount()
         drain(c, 1);
     }
 
+    public void testFlow() throws IOException
+    {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        declareBindConsume(c);
+        fill(1);
+        drain(c, 1);
+        channel.flow(false);
+        fill(1);
+        drain(c, 0);
+        channel.flow(true);
+        drain(c, 1);
+    }
+    
     protected void runLimitTests(int limit,
                                  boolean multiAck,
                                  boolean txMode,