QPID-5969 : [Java Common] Add support of AMQP 0-9-1 field-array type

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1616363 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 253 insertions, 15 deletions

diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQType.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQType.java
index 4a6cfe6077..b681e782a3 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQType.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQType.java
@@ -24,6 +24,7 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.util.Collection;
 
 /**
  * AMQType is a type that represents the different possible AMQP field table types. It provides operations for each
@@ -281,7 +282,63 @@ public enum AMQType
             }
         }
     },
-
+    /**
+     * Implements the field table type. The native value of a field table type will be an instance of
+     * {@link FieldTable}, which itself may contain name/value pairs encoded as {@link AMQTypedValue}s.
+     */
+    FIELD_ARRAY('A')
+            {
+                public int getEncodingSize(Object value)
+                {
+                    if (!(value instanceof Collection))
+                    {
+                        throw new IllegalArgumentException("Value is not a Collection.");
+                    }
+
+                    FieldArray fieldArrayValue = FieldArray.asFieldArray((Collection)value);
+
+                    return 4 + fieldArrayValue.getEncodingSize();
+                }
+
+                public Object toNativeValue(Object value)
+                {
+                    // Ensure that the value is a FieldTable.
+                    if (!(value instanceof Collection))
+                    {
+                        throw new IllegalArgumentException("Value cannot be converted to a FieldArray.");
+                    }
+
+                    return FieldArray.asFieldArray((Collection)value);
+                }
+
+                public void writeValueImpl(Object value, DataOutput buffer) throws IOException
+                {
+
+                    if (!(value instanceof FieldArray))
+                    {
+                        throw new IllegalArgumentException("Value is not a FieldArray.");
+                    }
+
+                    FieldArray fieldArrayValue = (FieldArray) value;
+
+                    // Loop over all name/values writing out into buffer.
+                    fieldArrayValue.writeToBuffer(buffer);
+                }
+
+                /**
+                 * Reads an instance of the type from a specified byte buffer.
+                 *
+                 * @param buffer The byte buffer to write it to.
+                 *
+                 * @return An instance of the type.
+                 */
+                public Object readValueFromBuffer(DataInput buffer) throws IOException
+                {
+                    // Read size of field table then all name/value pairs.
+                    return FieldArray.readFromBuffer(buffer);
+
+                }
+            },
     VOID('V')
     {
         public int getEncodingSize(Object value)
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
index c4dc86bf11..42a82eccba 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
@@ -24,6 +24,7 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.util.Collection;
 import java.util.Date;
 import java.util.Map;
 
@@ -271,7 +272,7 @@ public abstract class AMQTypedValue
         Class klass = val.getClass();
         if(klass == String.class)
         {
-            return AMQType.ASCII_STRING.asTypedValue(val);
+            return AMQType.LONG_STRING.asTypedValue(val);
         }
         else if(klass == Character.class)
         {
@@ -317,6 +318,14 @@ public abstract class AMQTypedValue
         {
             return AMQType.FIELD_TABLE.asTypedValue(FieldTable.convertToFieldTable((Map)val));
         }
-        return null;
+        else if(klass == FieldTable.class)
+        {
+            return AMQType.FIELD_TABLE.asTypedValue(val);
+        }
+        else if(val instanceof Collection)
+        {
+            return AMQType.FIELD_ARRAY.asTypedValue(val);
+        }
+        throw new IllegalArgumentException("Cannot convert an object of class " + val.getClass().getName() + " to an AMQP typed value");
     }
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldArray.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldArray.java
new file mode 100644
index 0000000000..d61d0e8385
--- /dev/null
+++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldArray.java
@@ -0,0 +1,130 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.framing;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.AbstractCollection;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+public class FieldArray<T> extends AbstractCollection<T>
+{
+
+    private final Collection<T> _underlying;
+
+    private FieldArray(final Collection<T> underlying)
+    {
+        _underlying = underlying;
+    }
+
+    @Override
+    public Iterator<T> iterator()
+    {
+        return _underlying.iterator();
+    }
+
+    @Override
+    public int size()
+    {
+        return _underlying.size();
+    }
+
+    public int getEncodingSize()
+    {
+        int size = 0;
+        for( T obj : this)
+        {
+            size += AMQTypedValue.toTypedValue(obj).getEncodingSize()+1;
+        }
+        return size;
+    }
+
+    public static <T> FieldArray<T>  asFieldArray(Collection<T> collection)
+    {
+        if(collection instanceof FieldArray)
+        {
+            return (FieldArray<T>) collection;
+        }
+        else
+        {
+            validateCollection(collection);
+            return new FieldArray<>(collection);
+        }
+    }
+
+    private static final Set<Class<?>> SUPPORTED_CLASSES = new HashSet<>(Arrays.asList(Boolean.class,
+                                                                                       Byte.class,
+                                                                                       Short.class,
+                                                                                       Character.class,
+                                                                                       Integer.class,
+                                                                                       Long.class,
+                                                                                       Float.class,
+                                                                                       Double.class,
+                                                                                       String.class,
+                                                                                       FieldTable.class,
+                                                                                       Date.class,
+                                                                                       BigDecimal.class,
+                                                                                       byte[].class));
+
+    private static <T> void validateCollection(final Collection<T> collection)
+    {
+        for(T val : collection)
+        {
+            if(!(val == null || SUPPORTED_CLASSES.contains(val.getClass()) || val instanceof Collection || val instanceof Map))
+            {
+                throw new IllegalArgumentException("Cannot convert an object of type " + val.getClass().getName());
+            }
+
+        }
+    }
+
+    public void writeToBuffer(final DataOutput buffer) throws IOException
+    {
+        buffer.writeInt(getEncodingSize());
+        for( T obj : this)
+        {
+            AMQTypedValue.toTypedValue(obj).writeToBuffer(buffer);
+        }
+    }
+
+    public static FieldArray<?> readFromBuffer(final DataInput buffer) throws IOException
+    {
+        ArrayList<Object> result = new ArrayList<>();
+        int size = EncodingUtils.readInteger(buffer);
+        byte[] data = new byte[size];
+        buffer.readFully(data);
+        ByteArrayDataInput slicedBuffer = new ByteArrayDataInput(data);
+        while(slicedBuffer.available() > 0)
+        {
+            result.add(AMQTypedValue.readFromBuffer(slicedBuffer).getValue());
+        }
+        return new FieldArray<>(result);
+    }
+}
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
index b9ed1b2154..588f33d755 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
@@ -20,18 +20,15 @@
  */
 package org.apache.qpid.framing;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.qpid.AMQPInvalidClassException;
-
 import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.Date;
 import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -40,6 +37,11 @@ import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.qpid.AMQPInvalidClassException;
+
 // extends FieldTable
 public class FieldTable
 {
@@ -577,6 +579,15 @@ public class FieldTable
         return setProperty(string, AMQType.ASCII_CHARACTER.asTypedValue(c));
     }
 
+    public Object setFieldArray(String string, Collection<?> collection)
+    {
+        return setFieldArray(AMQShortString.valueOf(string), collection);
+    }
+    public Object setFieldArray(AMQShortString string, Collection<?> collection)
+    {
+        return setProperty(string, AMQType.FIELD_ARRAY.asTypedValue(collection));
+    }
+
     public Object setBytes(String string, byte[] b)
     {
         return setBytes(AMQShortString.valueOf(string), b);
@@ -614,12 +625,12 @@ public class FieldTable
     {
         if (decimal.longValue() > Integer.MAX_VALUE)
         {
-            throw new UnsupportedOperationException("AMQP doesnot support decimals larger than " + Integer.MAX_VALUE);
+            throw new UnsupportedOperationException("AMQP does not support decimals larger than " + Integer.MAX_VALUE);
         }
 
         if (decimal.scale() > Byte.MAX_VALUE)
         {
-            throw new UnsupportedOperationException("AMQP doesnot support decimal scales larger than " + Byte.MAX_VALUE);
+            throw new UnsupportedOperationException("AMQP does not support decimal scales larger than " + Byte.MAX_VALUE);
         }
 
         return setProperty(string, AMQType.DECIMAL.asTypedValue(decimal));
@@ -694,6 +705,26 @@ public class FieldTable
         {
             return setChar(string, (Character) object);
         }
+        else if (object instanceof Collection)
+        {
+            return setFieldArray(string, (Collection)object);
+        }
+        else if (object instanceof FieldTable)
+        {
+            return setFieldTable(string, (FieldTable) object);
+        }
+        else if (object instanceof Map)
+        {
+            return setFieldTable(string, FieldTable.convertToFieldTable((Map) object));
+        }
+        else if (object instanceof Date)
+        {
+            return setTimestamp(string, ((Date) object).getTime());
+        }
+        else if (object instanceof BigDecimal)
+        {
+            return setDecimal(string, (BigDecimal) object);
+        }
         else if (object instanceof byte[])
         {
             return setBytes(string, (byte[]) object);
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/framing/FieldTableTest.java b/qpid/java/common/src/test/java/org/apache/qpid/framing/FieldTableTest.java
index 4b9ac81324..5c05adf997 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/framing/FieldTableTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/framing/FieldTableTest.java
@@ -20,16 +20,20 @@
  */
 package org.apache.qpid.framing;
 
-import org.junit.Assert;
-import junit.framework.TestCase;
-
-import org.apache.qpid.AMQPInvalidClassException;
-
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+
+import junit.framework.TestCase;
+import org.junit.Assert;
+
+import org.apache.qpid.AMQPInvalidClassException;
 
 public class FieldTableTest extends TestCase
 {
@@ -458,6 +462,7 @@ public class FieldTableTest extends TestCase
         innerTable.setShort("short", Short.MAX_VALUE);
         innerTable.setString("string", "hello");
         innerTable.setString("null-string", null);
+        innerTable.setFieldArray("field-array",Arrays.asList("hello",Integer.valueOf(42), Collections.emptyList()));
 
         // Put the inner table in the outer one.
         outerTable.setFieldTable("innerTable", innerTable);
@@ -487,6 +492,12 @@ public class FieldTableTest extends TestCase
             Assert.assertEquals(Short.valueOf(Short.MAX_VALUE), extractedTable.getShort("short"));
             Assert.assertEquals("hello", extractedTable.getString("string"));
             Assert.assertNull(extractedTable.getString("null-string"));
+            Collection fieldArray = (Collection) extractedTable.get("field-array");
+            Assert.assertEquals(3, fieldArray.size());
+            Iterator iter = fieldArray.iterator();
+            assertEquals("hello",iter.next());
+            assertEquals(Integer.valueOf(42), iter.next());
+            assertTrue(((Collection)iter.next()).isEmpty());
         }
         catch (AMQFrameDecodingException e)
         {