merge default into bug20337

Author: Alexandru Scvortov

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

@@ -297,6 +297,9 @@ else if(value instanceof byte[]) {
         else if(value instanceof List) {
             acc += 4 + arraySize((List<?>)value);
         }
+        else if(value instanceof Object[]) {
+            acc += 4 + arraySize((Object[])value);
+        }
         else if(value == null) {
         }
         else {
@@ -305,7 +308,7 @@ else if(value == null) {
         return acc;
     }
 
-    /** Computes the AMQP wire-protocol length of an encoded field-array */
+    /** Computes the AMQP wire-protocol length of an encoded field-array of type List<?> */
     public static long arraySize(List<?> values)
         throws UnsupportedEncodingException
     {
@@ -315,7 +318,16 @@ public static long arraySize(List<?> values)
         }
         return acc;
     }
-  
+
+    /** Computes the AMQP wire-protocol length of an encoded field-array of type Object[] */
+    public static long arraySize(Object[] values) throws UnsupportedEncodingException {
+        long acc = 0;
+        for (Object value : values) {
+            acc += fieldValueSize(value);
+        }
+        return acc;
+    }
+
     /** Computes the AMQP wire-protocol length of a protocol-encoded long string. */
     private static int longStrSize(String str)
         throws UnsupportedEncodingException

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

@@ -200,6 +200,10 @@ else if(value instanceof List) {
             writeOctet('A');
             writeArray((List<?>)value);
         }
+        else if(value instanceof Object[]) {
+            writeOctet('A');
+            writeArray((Object[])value);
+        }
         else {
             throw new IllegalArgumentException
                 ("Invalid value type: " + value.getClass().getName());
@@ -220,6 +224,20 @@ public final void writeArray(List<?> value)
         }
     }
 
+    public final void writeArray(Object[] value)
+        throws IOException
+    {
+        if (value==null) {
+            out.write(0);
+        }
+        else {
+            out.writeInt((int)Frame.arraySize(value));
+            for (Object item : value) {
+                writeFieldValue(item);
+            }
+        }
+    }
+
     /** Public API - encodes an octet from an int. */
     public final void writeOctet(int octet)
         throws IOException

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

@@ -18,14 +18,17 @@
 package com.rabbitmq.client.test.functional;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.test.BrokerTestCase;
 
-// Test queue auto-delete and exclusive semantics.
+/**
+ * Test queue auto-delete and exclusive semantics.
+ */
 public class QueueLifecycle extends BrokerTestCase {
 
     void verifyQueueExists(String name) throws IOException {
@@ -77,12 +80,13 @@ void verifyNotEquivalent(boolean durable, boolean exclusive,
         fail("Queue.declare should have been rejected as not equivalent");
     }
 
-    // From amqp-0-9-1.xml, for "passive" property, "equivalent" rule:
-    // "If not set and the queue exists, the server MUST check that the
-    // existing queue has the same values for durable, exclusive,
-    // auto-delete, and arguments fields. The server MUST respond with
-    // Declare-Ok if the requested queue matches these fields, and MUST
-    // raise a channel exception if not."
+    /** From amqp-0-9-1.xml, for "passive" property, "equivalent" rule:
+     * "If not set and the queue exists, the server MUST check that the
+     * existing queue has the same values for durable, exclusive,
+     * auto-delete, and arguments fields. The server MUST respond with
+     * Declare-Ok if the requested queue matches these fields, and MUST
+     * raise a channel exception if not."
+     */
     public void testQueueEquivalence() throws IOException {
         String q = "queue";
         channel.queueDeclare(q, false, false, false, null);
@@ -154,4 +158,12 @@ public void testExclusiveGoesWithConnection() throws IOException {
         verifyQueueMissing(name);
     }
 
+    public void testArgumentArrays() throws IOException {
+        Map<String, Object> args = new HashMap<String, Object>();
+        String[] arr = new String[]{"foo", "bar", "baz"};
+        args.put("my-key", arr);
+        String queueName = "argumentArraysQueue";
+        channel.queueDeclare(queueName, true, true, false, args);
+        verifyQueue(queueName, true, true, false, args);
+    }
 }

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

@@ -329,4 +329,127 @@ public void testShuffleAcksBeforeRollback()
         basicAck(tags[2], false);
         txCommit();
     }
+
+    private abstract class NackMethod {
+        abstract public void nack(long tag, boolean requeue)
+            throws IOException;
+
+        public void nack(boolean requeue)
+            throws IOException
+        {
+            nack(latestTag, requeue);
+        }
+
+        public void nack()
+            throws IOException
+        {
+            nack(latestTag, true);
+        }
+    }
+
+    private NackMethod basicNack = new NackMethod() {
+            public void nack(long tag, boolean requeue)
+                throws IOException
+            {
+                channel.basicNack(tag, false, requeue);
+            }
+        };
+
+    private NackMethod basicReject = new NackMethod() {
+            public void nack(long tag, boolean requeue)
+                throws IOException
+            {
+                channel.basicReject(tag, requeue);
+            }
+        };
+
+    /*
+      messages with nacks get requeued after the transaction commit.
+      messages with nacks with requeue = false are not requeued.
+    */
+    public void commitNacks(NackMethod method)
+        throws IOException
+    {
+        basicPublish();
+        basicPublish();
+        txSelect();
+        basicGet();
+        method.nack();
+        basicGet();
+        method.nack(false);
+        assertNull(basicGet());
+        txCommit();
+        assertNotNull(basicGet());
+        assertNull(basicGet());
+    }
+
+    public void rollbackNacks(NackMethod method)
+        throws IOException
+    {
+        basicPublish();
+        txSelect();
+        basicGet();
+        method.nack(true);
+        txRollback();
+        assertNull(basicGet());
+    }
+
+    public void commitAcksAndNacks(NackMethod method)
+        throws IOException
+    {
+        for (int i = 0; i < 3; i++) {
+            basicPublish();
+        }
+        txSelect();
+        long tags[] = new long[3];
+        for (int i = 0; i < 3; i++) {
+            tags[i] = basicGet().getEnvelope().getDeliveryTag();
+        }
+        basicAck(tags[1], false);
+        basicAck(tags[0], false);
+        method.nack(tags[2], false);
+        txRollback();
+        basicAck(tags[2], false);
+        method.nack(tags[0], true);
+        method.nack(tags[1], false);
+        txCommit();
+        assertNotNull(basicGet());
+        assertNull(basicGet());
+    }
+
+    public void testCommitNacks()
+        throws IOException
+    {
+        commitNacks(basicNack);
+    }
+
+    public void testRollbackNacks()
+        throws IOException
+    {
+        rollbackNacks(basicNack);
+    }
+
+    public void testCommitAcksAndNacks()
+        throws IOException
+    {
+        commitAcksAndNacks(basicNack);
+    }
+
+    public void testCommitRejects()
+        throws IOException
+    {
+        commitNacks(basicReject);
+    }
+
+    public void testRollbackRejects()
+        throws IOException
+    {
+        rollbackNacks(basicReject);
+    }
+
+    public void testCommitAcksAndRejects()
+        throws IOException
+    {
+        commitAcksAndNacks(basicReject);
+    }
 }