merge default into bug14587

Author: michaelklishin

src/com/rabbitmq/client/Channel.java

@@ -26,7 +26,6 @@
 import com.rabbitmq.client.AMQP.Tx;
 import com.rabbitmq.client.AMQP.Basic;
 import com.rabbitmq.client.AMQP.Confirm;
-import com.rabbitmq.client.AMQP.Channel.FlowOk;
 
 /**
  * Public API: Interface to an AMQ channel. See the <a href="http://www.amqp.org/">spec</a> for details.
@@ -90,17 +89,11 @@ public interface Channel extends ShutdownNotifier {
     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;
-
-    /**
-     * Return the current Channel.Flow settings.
+     * Indicates whether the server has asked this client to stop
+     * sending content-bearing commands (such as basic.publish) by
+     * issueing a channel.flow{active=false}.
      */
-    FlowOk getFlow();
+    boolean flowBlocked();
 
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code

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

@@ -1061,13 +1061,8 @@ public Confirm.SelectOk confirmSelect()
     }
 
     /** Public API - {@inheritDoc} */
-    public Channel.FlowOk flow(final boolean a) throws IOException {
-        return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow(a)).getMethod();
-    }
-
-    /** Public API - {@inheritDoc} */
-    public Channel.FlowOk getFlow() {
-        return new Channel.FlowOk(!_blockContent);
+    public boolean flowBlocked() {
+        return _blockContent;
     }
 
     /** Public API - {@inheritDoc} */

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

@@ -344,43 +344,6 @@ 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);
-    }
-
-    public void testLimitAndFlow() throws IOException
-    {
-        channel.basicQos(1);
-        QueueingConsumer c = new QueueingConsumer(channel);
-        declareBindConsume(c);
-        channel.flow(false);
-        fill(3);
-        drain(c, 0);
-        channel.flow(true);
-        ack(drain(c, 1), false);
-        drain(c, 1);
-        channel.basicQos(0);
-        drain(c, 1);
-    }
-
-    public void testNoConsumers() throws Exception {
-        String q = declareBind(channel);
-        fill(1);
-        channel.flow(false);
-        QueueingConsumer c = new QueueingConsumer(channel);
-        channel.basicConsume(q, c);
-        drain(c, 0);
-    }
-
     public void testRecoverReducesLimit() throws Exception {
         channel.basicQos(2);
         QueueingConsumer c = new QueueingConsumer(channel);

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

@@ -17,8 +17,8 @@
 package com.rabbitmq.client.test.functional;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.test.BrokerTestCase;
-import com.rabbitmq.tools.Host;
 
 import java.io.IOException;
 
@@ -36,18 +36,8 @@ public void testInvalidUserId() {
             fail("Accepted publish with incorrect user ID");
         } catch (IOException e) {
             checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
-        }
-    }
-
-    public void testImpersonatedUserId() throws IOException {
-        Host.rabbitmqctl("set_user_tags guest administrator impersonator");
-        connection = null;
-        channel = null;
-        setUp();
-        try {
-            publish(BAD);
-        } finally {
-            Host.rabbitmqctl("set_user_tags guest administrator");
+        } catch (AlreadyClosedException e) {
+            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
         }
     }
 

test/src/com/rabbitmq/client/test/performance/QosScaling.java

@@ -129,9 +129,7 @@ public long run() throws IOException {
         channel.basicQos(1);
         QueueingConsumer consumer = new QueueingConsumer(channel);
         try {
-            channel.flow(false);
             publish(consume(consumer));
-            channel.flow(true);
             return drain(consumer);
         } finally {
             connection.abort();

test/src/com/rabbitmq/examples/PerfTest.java

@@ -70,8 +70,6 @@ public static void main(String[] args) {
 
             String uri           = strArg(cmd, 'h', "amqp://localhost");
 
-            boolean exclusive  = "".equals(queueName);
-
             //setup
             PrintlnStats stats = new PrintlnStats(1000L * samplingInterval,
                                     producerCount > 0,
@@ -88,14 +86,13 @@ public static void main(String[] args) {
 
             MulticastParams p = new MulticastParams();
             p.setAutoAck(          autoAck);
-            p.setAutoDelete(       !exclusive);
+            p.setAutoDelete(       true);
             p.setConfirm(          confirm);
             p.setConsumerCount(    consumerCount);
             p.setConsumerMsgCount( consumerMsgCount);
             p.setConsumerTxSize(   consumerTxSize);
             p.setExchangeName(     exchangeName);
             p.setExchangeType(     exchangeType);
-            p.setExclusive(        exclusive);
             p.setFlags(            flags);
             p.setMultiAckEvery(    multiAckEvery);
             p.setMinMsgSize(       minMsgSize);

test/src/com/rabbitmq/examples/perf/MulticastParams.java

@@ -49,7 +49,6 @@ public class MulticastParams {
 
     private int multiAckEvery = 0;
     private boolean autoAck = true;
-    private boolean exclusive = true;
     private boolean autoDelete = false;
 
     private boolean predeclared;
@@ -131,10 +130,6 @@ public void setFlags(List<?> flags) {
         this.flags = flags;
     }
 
-    public void setExclusive(boolean exclusive) {
-        this.exclusive = exclusive;
-    }
-
     public void setAutoDelete(boolean autoDelete) {
         this.autoDelete = autoDelete;
     }
@@ -187,7 +182,7 @@ public Consumer createConsumer(Connection connection, Stats stats, String id) th
     }
 
     public boolean shouldConfigureQueue() {
-        return consumerCount == 0 && !queueName.equals("") && !exclusive;
+        return consumerCount == 0 && !queueName.equals("");
     }
 
     public String configureQueue(Connection connection, String id) throws IOException {
@@ -199,7 +194,7 @@ public String configureQueue(Connection connection, String id) throws IOExceptio
         if (!predeclared || !queueExists(connection, queueName)) {
             qName = channel.queueDeclare(queueName,
                                          flags.contains("persistent"),
-                                         exclusive, autoDelete,
+                                         false, autoDelete,
                                          null).getQueue();
         }
         channel.queueBind(qName, exchangeName, id);