Merged 18557 into default

Author: 0x6e6562

src/com/rabbitmq/client/Channel.java

@@ -120,6 +120,35 @@ public interface Channel extends ShutdownNotifier{
      */
     void setReturnListener(ReturnListener listener);
 
+    /**
+     * Request specific "quality of service" settings.
+     *
+     * These settings impose limits on the amount of data the server
+     * will deliver to consumers before requiring the receipt of
+     * acknowledgements.
+     * Thus they provide a means of consumer-initiated flow control.
+     * @see com.rabbitmq.client.AMQP.Basic.Qos
+     * @param prefetchSize maximum amount of content (measured in
+     * octets) that the server will deliver, 0 if unlimited
+     * @param prefetchCount maximum number of messages that the server
+     * will deliver, 0 if unlimited
+     * @param global true if the settings should be applied to the
+     * entire connection rather than just the current channel
+     * @throws java.io.IOException if an error is encountered
+     */
+    void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
+
+    /**
+     * Request a specific prefetchCount "quality of service" settings
+     * for this channel.
+     *
+     * @see #basicQos(int, int, boolean)
+     * @param prefetchCount maximum number of messages that the server
+     * will deliver, 0 if unlimited
+     * @throws java.io.IOException if an error is encountered
+     */
+    void basicQos(int prefetchCount) throws IOException;
+
     /**
      * Publish a message with both "mandatory" and "immediate" flags set to false
      * @see com.rabbitmq.client.AMQP.Basic.Publish

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

@@ -360,6 +360,18 @@ public void close(int closeCode,
         }
     }    
 
+    public void basicQos(int prefetchSize, int prefetchCount, boolean global)
+	throws IOException
+    {
+	exnWrappingRpc(new Basic.Qos(prefetchSize, prefetchCount, global));
+    }
+
+    public void basicQos(int prefetchCount)
+	throws IOException
+    {
+	basicQos(0, prefetchCount, false);
+    }
+
     /**
      * Public API - Publish a message with both "mandatory" and
      * "immediate" flags set to false

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

@@ -36,8 +36,12 @@
 import junit.framework.TestCase;
 
 import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.ShutdownSignalException;
+
+import com.rabbitmq.client.AMQP;
 
 public class BrokerTestCase extends TestCase
 {
@@ -120,4 +124,17 @@ public void closeChannel()
         }
     }
 
+    public void checkShutdownSignal(int expectedCode, IOException ioe) {
+        ShutdownSignalException sse = (ShutdownSignalException) ioe.getCause();
+        Command closeCommand = (Command) sse.getReason();
+        channel = null;
+        if (sse.isHardError()) {
+            connection = null;
+            AMQP.Connection.Close closeMethod = (AMQP.Connection.Close) closeCommand.getMethod();
+            assertEquals(expectedCode, closeMethod.getReplyCode());
+        } else {
+            AMQP.Channel.Close closeMethod = (AMQP.Channel.Close) closeCommand.getMethod();
+            assertEquals(expectedCode, closeMethod.getReplyCode());
+        }
+    }
 }

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

@@ -45,6 +45,7 @@ public static TestSuite suite() {
         suite.addTestSuite(RequeueOnChannelClose.class);
         suite.addTestSuite(DurableOnTransient.class);
         suite.addTestSuite(NoRequeueOnCancel.class);
+        suite.addTestSuite(QosTests.class);
         return suite;
     }
 }