merge with default

Author: Rob Harrop

src/com/rabbitmq/client/Channel.java

@@ -412,26 +412,15 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      */
     Queue.UnbindOk queueUnbind(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException;
 
-    /**
-     * Purges the contents of the given queue and awaits a completion.
-     * @see com.rabbitmq.client.AMQP.Queue.Purge
-     * @see com.rabbitmq.client.AMQP.Queue.PurgeOk
-     * @param queue the name of the queue
-     * @return a purge-confirm method if the purge was executed succesfully
-     * @throws java.io.IOException if an error is encountered
-     */
-    Queue.PurgeOk queuePurge(String queue) throws IOException;
-
     /**
      * Purges the contents of the given queue.
      * @see com.rabbitmq.client.AMQP.Queue.Purge
      * @see com.rabbitmq.client.AMQP.Queue.PurgeOk
      * @param queue the name of the queue
-     * @param nowait whether to await completion of the purge
      * @return a purge-confirm method if the purge was executed succesfully
      * @throws java.io.IOException if an error is encountered
      */
-    Queue.PurgeOk queuePurge(String queue, boolean nowait) throws IOException;
+    Queue.PurgeOk queuePurge(String queue) throws IOException;
 
     /**
      * Retrieve a message from a queue using {@link com.rabbitmq.client.AMQP.Basic.Get}
@@ -519,12 +508,13 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * @param noLocal flag set to true unless server local buffering is required
      * @param exclusive true if this is an exclusive consumer
      * @param callback an interface to the consumer object
+     * @param arguments a set of arguments for the consume
      * @return the consumerTag associated with the new consumer
      * @throws java.io.IOException if an error is encountered
      * @see com.rabbitmq.client.AMQP.Basic.Consume
      * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
      */
-    String basicConsume(String queue, boolean noAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> filter, Consumer callback) throws IOException;
+    String basicConsume(String queue, boolean noAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException;
 
     /**
      * Cancel a consumer. Calls the consumer's {@link Consumer#handleCancelOk}

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

@@ -596,16 +596,9 @@ public Queue.UnbindOk queueUnbind(String queue, String exchange, String routingK
     /** Public API - {@inheritDoc} */
     public Queue.PurgeOk queuePurge(String queue)
         throws IOException
-    {
-        return queuePurge(queue, false);
-    }
-
-    /** Public API - {@inheritDoc} */
-    public Queue.PurgeOk queuePurge(String queue, boolean nowait)
-        throws IOException
     {
         return (Queue.PurgeOk)
-            exnWrappingRpc(new Queue.Purge(TICKET, queue, nowait)).getMethod();
+            exnWrappingRpc(new Queue.Purge(TICKET, queue, false)).getMethod();
     }
 
     /** Public API - {@inheritDoc} */
@@ -677,7 +670,7 @@ public String basicConsume(String queue, boolean noAck, String consumerTag,
 
     /** Public API - {@inheritDoc} */
     public String basicConsume(String queue, boolean noAck, String consumerTag,
-                               boolean noLocal, boolean exclusive, Map<String, Object> filter,
+                               boolean noLocal, boolean exclusive, Map<String, Object> arguments,
                                final Consumer callback)
         throws IOException
     {
@@ -702,7 +695,7 @@ public String transformReply(AMQCommand replyCommand) {
 
         rpc(new Basic.Consume(TICKET, queue, consumerTag,
                               noLocal, noAck, exclusive,
-                              false, filter),
+                              false, arguments),
             k);
 
         try {

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

@@ -46,7 +46,7 @@ public class QueueLease extends BrokerTestCase {
 
     // Currently the expiration timer is very responsive but this may
     // very well change in the future, so tweak accordingly.
-    private final static long QUEUE_EXPIRES = 1000L; // msecs
+    private final static int QUEUE_EXPIRES = 1000; // msecs
     private final static int SHOULD_EXPIRE_WITHIN = 2000;
 
     /**
@@ -65,27 +65,42 @@ public void testDoesNotExpireOthers() throws IOException,
         verifyQueueExpires(TEST_NORMAL_QUEUE, false);
     }
 
-    /**
-     * Verify that the server throws an error if the type of x-expires is not
-     * long.
-     */
-    public void testExpireMustBeLong() throws IOException {
+    public void testExpireMayBeByte() throws IOException {
         Map<String, Object> args = new HashMap<String, Object>();
-        args.put("x-expires", 100);
+        args.put("x-expires", (byte)100);
 
         try {
-            channel
-                    .queueDeclare("expiresMustBeLong", false, false, false,
-                            args);
-            fail("server accepted x-expires not of type long");
+            channel.queueDeclare("expiresMayBeByte", false, true, false, args);
         } catch (IOException e) {
-            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
+            fail("server did not accept x-expires of type byte");
+        }
+    }
+
+    public void testExpireMayBeShort() throws IOException {
+        Map<String, Object> args = new HashMap<String, Object>();
+        args.put("x-expires", (short)100);
+
+        try {
+            channel.queueDeclare("expiresMayBeShort", false, true, false, args);
+        } catch (IOException e) {
+            fail("server did not accept x-expires of type short");
+        }
+    }
+
+    public void testExpireMayBeLong() throws IOException {
+        Map<String, Object> args = new HashMap<String, Object>();
+        args.put("x-expires", 100L);
+
+        try {
+            channel.queueDeclare("expiresMayBeLong", false, true, false, args);
+        } catch (IOException e) {
+            fail("server did not accept x-expires of type long");
         }
     }
 
     public void testExpireMustBeGtZero() throws IOException {
         Map<String, Object> args = new HashMap<String, Object>();
-        args.put("x-expires", 0L);
+        args.put("x-expires", 0);
 
         try {
             channel.queueDeclare("expiresMustBeGtZero", false, false, false,
@@ -98,7 +113,7 @@ public void testExpireMustBeGtZero() throws IOException {
 
     public void testExpireMustBePositive() throws IOException {
         Map<String, Object> args = new HashMap<String, Object>();
-        args.put("x-expires", -10L);
+        args.put("x-expires", -10);
 
         try {
             channel.queueDeclare("expiresMustBePositive", false, false, false,
@@ -115,9 +130,9 @@ public void testExpireMustBePositive() throws IOException {
      */
     public void testQueueRedeclareEquivalence() throws IOException {
         Map<String, Object> args1 = new HashMap<String, Object>();
-        args1.put("x-expires", 10000L);
+        args1.put("x-expires", 10000);
         Map<String, Object> args2 = new HashMap<String, Object>();
-        args2.put("x-expires", 20000L);
+        args2.put("x-expires", 20000);
 
         channel.queueDeclare(TEST_EXPIRE_REDECLARE_QUEUE, false, false, false,
                 args1);
@@ -145,6 +160,7 @@ void verifyQueueExpires(String name, boolean expire) throws IOException,
         try {
             channel.queueDeclarePassive(name);
         } catch (IOException e) {
+            checkShutdownSignal(AMQP.NOT_FOUND, e);
             fail("Queue expired before deadline.");
         }
 

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

@@ -13,13 +13,53 @@
  * Test that the server correctly handles us when we send it bad frames
  */
 public class UnexpectedFrames extends BrokerTestCase {
+
     private interface Confuser {
         public Frame confuse(Frame frame) throws IOException;
     }
 
-    @Override protected void setUp() throws IOException {}
+    private static class ConfusedFrameHandler extends SocketFrameHandler {
+
+        private boolean confusedOnce = false;
+
+        public ConfusedFrameHandler(Socket socket) throws IOException {
+            super(socket);
+        }
+
+        @Override
+        public void writeFrame(Frame frame) throws IOException {
+            if (confusedOnce) {
+                super.writeFrame(frame);
+            } else {
+                Frame confusedFrame = confuser.confuse(frame);
+                if (confusedFrame != frame) confusedOnce = true;
+                if (confusedFrame != null) {
+                    super.writeFrame(confusedFrame);
+                }
+            }
+        }
+
+        public Confuser confuser = new Confuser() {
+            public Frame confuse(Frame frame) {
+                // Do nothing to start with, we need to negotiate before the
+                // server will send us unexpected_frame errors
+                return frame;
+            }
+        };
+    }
 
-    @Override protected void tearDown() throws IOException {}
+    private static class ConfusedConnectionFactory extends ConnectionFactory {
+
+        @Override protected FrameHandler createFrameHandler(Socket sock)
+            throws IOException {
+            return new ConfusedFrameHandler(sock);
+        }
+    }
+
+    public UnexpectedFrames() {
+        super();
+        connectionFactory = new ConfusedConnectionFactory();
+    }
 
     public void testMissingHeader() throws IOException {
         expectUnexpectedFrameError(new Confuser() {
@@ -38,7 +78,11 @@ public Frame confuse(Frame frame) {
                 if (frame.type == AMQP.FRAME_METHOD) {
                     // We can't just skip the method as that will lead us to
                     // send 0 bytes and hang waiting for a response.
-                    frame.type = AMQP.FRAME_HEADER;
+                    Frame confusedFrame = new Frame(AMQP.FRAME_HEADER,
+                                                    frame.channel,
+                                                    frame.payload);
+                    confusedFrame.accumulator = frame.accumulator;
+                    return confusedFrame;
                 }
                 return frame;
             }
@@ -73,53 +117,16 @@ public Frame confuse(Frame frame) throws IOException {
         });
     }
 
-    private void expectUnexpectedFrameError(Confuser confuser) throws IOException {
-        ConnectionFactory factory = new ConnectionFactory();
-        Socket socket = factory.getSocketFactory().createSocket("localhost",
-                AMQP.PROTOCOL.PORT);
-        ConfusedFrameHandler handler = new ConfusedFrameHandler(socket);
-        AMQConnection connection = new AMQConnection(factory, handler);
-        connection.start();
-        Channel channel = connection.createChannel();
-
-        handler.confuser = confuser;
-
-        try {
-            //NB: the frame confuser relies on the encoding of the
-            //method field to be at least 8 bytes long
-            channel.basicPublish("", "routing key", null, "Hello".getBytes());
-            channel.basicQos(0);
-            fail("We should have seen an UNEXPECTED_FRAME by now");
-        }
-        catch (IOException e) {
-            checkShutdownSignal(AMQP.UNEXPECTED_FRAME, e);
-        }
-    }
+    private void expectUnexpectedFrameError(Confuser confuser)
+        throws IOException {
 
-    private static class ConfusedFrameHandler extends SocketFrameHandler {
-        public ConfusedFrameHandler(Socket socket) throws IOException {
-            super(socket);
-        }
+        ((ConfusedFrameHandler)((AMQConnection)connection).getFrameHandler()).
+            confuser = confuser;
 
-        @Override
-        public void writeFrame(Frame frame) throws IOException {
-            Frame confusedFrame = new Frame();
-            confusedFrame.accumulator = frame.accumulator;
-            confusedFrame.channel = frame.channel;
-            confusedFrame.type = frame.type;
-
-            confusedFrame = confuser.confuse(confusedFrame);
-            if (confusedFrame != null) {
-                super.writeFrame(confusedFrame);
-            }
-        }
-
-        public Confuser confuser = new Confuser() {
-            public Frame confuse(Frame frame) {
-                // Do nothing to start with, we need to negotiate before the
-                // server will send us unexpected_frame errors
-                return frame;
-            }
-        };
+        //NB: the frame confuser relies on the encoding of the
+        //method field to be at least 8 bytes long
+        channel.basicPublish("", "routing key", null, "Hello".getBytes());
+        expectError(AMQP.UNEXPECTED_FRAME);
     }
+
 }