add API and test for basic.reject

Author: rade

src/com/rabbitmq/client/Channel.java

@@ -457,6 +457,17 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      */
     void basicAck(long deliveryTag, boolean multiple) throws IOException;
 
+    /**
+     * Reject a message. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
+     * or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
+     * containing the received message being rejected.
+     * @see com.rabbitmq.client.AMQP.Basic.Reject
+     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
+     * @param requeue true if the rejected message should be requeued rather than discarded
+     * @throws java.io.IOException if an error is encountered
+     */
+    void basicReject(long deliveryTag, boolean requeue) throws IOException;
+
     /**
      * Start a non-nolocal, non-exclusive consumer, with
      * explicit acknowledgements required and a server-generated consumerTag.

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

@@ -646,6 +646,13 @@ public void basicAck(long deliveryTag, boolean multiple)
         transmit(new Basic.Ack(deliveryTag, multiple));
     }
 
+    /** Public API - {@inheritDoc} */
+    public void basicReject(long deliveryTag, boolean requeue)
+        throws IOException
+    {
+        transmit(new Basic.Reject(deliveryTag, requeue));
+    }
+
     /** Public API - {@inheritDoc} */
     public String basicConsume(String queue, Consumer callback)
         throws IOException

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

@@ -43,6 +43,7 @@ public static TestSuite suite() {
         suite.addTestSuite(Routing.class);
         suite.addTestSuite(BindingLifecycle.class);
         suite.addTestSuite(Recover.class);
+        suite.addTestSuite(Reject.class);
         suite.addTestSuite(TransactionalRecover.class);
         suite.addTestSuite(Transactions.class);
         suite.addTestSuite(PersistentTransactions.class);

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

@@ -0,0 +1,76 @@
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.mozilla.org/MPL/
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is RabbitMQ.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.QueueingConsumer.Delivery;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+public class Reject extends BrokerTestCase
+{
+
+    protected long checkDelivery(Delivery d, byte[] msg, boolean redelivered)
+    {
+        assertNotNull(d);
+        assertTrue(Arrays.equals(msg, d.getBody()));
+        assertEquals(d.getEnvelope().isRedeliver(), redelivered);
+        return d.getEnvelope().getDeliveryTag();
+    }
+
+    public void testReject()
+        throws IOException, InterruptedException
+    {
+        String q = channel.queueDeclare("", false, true, false, null).getQueue();
+
+        QueueingConsumer c = new QueueingConsumer(channel);
+        String consumerTag = channel.basicConsume(q, false, c);
+
+        byte[] m1 = "1".getBytes();
+        byte[] m2 = "2".getBytes();
+
+        basicPublishVolatile(m1, q);
+        basicPublishVolatile(m2, q);
+
+        long tag1 = checkDelivery(c.nextDelivery(), m1, false);
+        long tag2 = checkDelivery(c.nextDelivery(), m2, false);
+        channel.basicReject(tag2, true);
+        long tag3 = checkDelivery(c.nextDelivery(), m2, true);
+        channel.basicCancel(consumerTag);
+        channel.basicReject(tag3, false);
+        assertNull(channel.basicGet(q, false));
+        channel.basicAck(tag1, false);
+    }
+}