PROTON-1708 Optimizations for the EncoderImpl and DecoderImpl
Provides several performance enhancements for the AMQP codec EncoderImpl
and DecoderImpl classes.
* Reduce instanceof and other type checks fro known types or by looking
ahead in the working buffer
* Reduce the number of throw away objects created during encode and
decode phase such as dynamic type constructor and char decoders etc.
* Provide fast write paths for both primitive types and known Described
types that are most often used
* Provide fast write path for most primitive write operations and for
Described types that are most often used
* Ensure the hot path for most read and write operations can be inlined
for the most common types.
* Adds some look ahead methods for users of the decoder that allow for
skipping types that aren't needed
* Allow Map types to be given fixed Type handlers for the Map keys for
those AMQP Map types whose Key types are specified to avoid the overhead
of looking them up on each Key read / write.
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/Symbol.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/Symbol.java
index 17e6177..4e3d476 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/Symbol.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/Symbol.java
@@ -21,16 +21,23 @@
package org.apache.qpid.proton.amqp;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
public final class Symbol implements Comparable<Symbol>, CharSequence
{
private final String _underlying;
+ private final byte[] _underlyingBytes;
+
private static final ConcurrentHashMap<String, Symbol> _symbols = new ConcurrentHashMap<String, Symbol>(2048);
private Symbol(String underlying)
{
_underlying = underlying;
+ _underlyingBytes = underlying.getBytes(StandardCharsets.US_ASCII);
}
public int length()
@@ -90,5 +97,13 @@
return symbol;
}
+ public void writeTo(WritableBuffer buffer)
+ {
+ buffer.put(_underlyingBytes, 0, _underlyingBytes.length);
+ }
+ public void writeTo(ByteBuffer buffer)
+ {
+ buffer.put(_underlyingBytes, 0, _underlyingBytes.length);
+ }
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/AMQPDefinedTypes.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/AMQPDefinedTypes.java
index 2e2f9e0..79dea5d 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/AMQPDefinedTypes.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/AMQPDefinedTypes.java
@@ -23,15 +23,6 @@
package org.apache.qpid.proton.codec;
-import org.apache.qpid.proton.amqp.transport.Attach;
-import org.apache.qpid.proton.amqp.transport.Begin;
-import org.apache.qpid.proton.amqp.transport.Close;
-import org.apache.qpid.proton.amqp.transport.Detach;
-import org.apache.qpid.proton.amqp.transport.Disposition;
-import org.apache.qpid.proton.amqp.transport.End;
-import org.apache.qpid.proton.amqp.transport.Flow;
-import org.apache.qpid.proton.amqp.transport.Open;
-import org.apache.qpid.proton.amqp.transport.Transfer;
import org.apache.qpid.proton.codec.messaging.*;
import org.apache.qpid.proton.codec.security.*;
import org.apache.qpid.proton.codec.transaction.*;
@@ -47,7 +38,6 @@
registerSecurityTypes(decoder, encoder);
}
-
public static void registerTransportTypes(Decoder decoder, EncoderImpl encoder)
{
OpenType.register(decoder, encoder);
@@ -60,21 +50,25 @@
EndType.register(decoder, encoder);
CloseType.register(decoder, encoder);
ErrorConditionType.register(decoder, encoder);
+
+ FastPathFlowType.register(decoder, encoder);
+ FastPathTransferType.register(decoder, encoder);
+ FastPathDispositionType.register(decoder, encoder);
}
public static void registerMessagingTypes(Decoder decoder, EncoderImpl encoder)
{
HeaderType.register(decoder, encoder);
+ AcceptedType.register(decoder , encoder);
+ PropertiesType.register( decoder, encoder );
DeliveryAnnotationsType.register(decoder, encoder);
MessageAnnotationsType.register(decoder, encoder);
- PropertiesType.register( decoder, encoder );
ApplicationPropertiesType.register(decoder, encoder);
DataType.register(decoder, encoder);
AmqpSequenceType.register(decoder, encoder);
AmqpValueType.register(decoder, encoder);
FooterType.register(decoder, encoder);
ReceivedType.register(decoder, encoder);
- AcceptedType.register(decoder , encoder);
RejectedType.register(decoder, encoder);
ReleasedType.register(decoder, encoder);
ModifiedType.register(decoder, encoder);
@@ -84,6 +78,17 @@
DeleteOnNoLinksType.register(decoder, encoder);
DeleteOnNoMessagesType.register(decoder, encoder);
DeleteOnNoLinksOrMessagesType.register(decoder, encoder);
+
+ FastPathHeaderType.register(decoder, encoder);
+ FastPathAcceptedType.register(decoder , encoder);
+ FastPathPropertiesType.register( decoder, encoder );
+ FastPathDeliveryAnnotationsType.register(decoder, encoder);
+ FastPathMessageAnnotationsType.register(decoder, encoder);
+ FastPathApplicationPropertiesType.register(decoder, encoder);
+ FastPathDataType.register(decoder, encoder);
+ FastPathAmqpSequenceType.register(decoder, encoder);
+ FastPathAmqpValueType.register(decoder, encoder);
+ FastPathFooterType.register(decoder, encoder);
}
public static void registerTransactionTypes(Decoder decoder, EncoderImpl encoder)
@@ -103,5 +108,4 @@
SaslResponseType.register(decoder, encoder);
SaslOutcomeType.register(decoder, encoder);
}
-
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/AbstractDescribedType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/AbstractDescribedType.java
index d31ac40..2f34fd5 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/AbstractDescribedType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/AbstractDescribedType.java
@@ -28,17 +28,27 @@
abstract public class AbstractDescribedType<T,M> implements AMQPType<T>
{
-
+ private final DecoderImpl _decoder;
private final EncoderImpl _encoder;
private final Map<TypeEncoding<M>, TypeEncoding<T>> _encodings = new HashMap<TypeEncoding<M>, TypeEncoding<T>>();
public AbstractDescribedType(EncoderImpl encoder)
{
_encoder = encoder;
+ _decoder = encoder.getDecoder();
}
abstract protected UnsignedLong getDescriptor();
+ public EncoderImpl getEncoder()
+ {
+ return _encoder;
+ }
+
+ public DecoderImpl getDecoder()
+ {
+ return _decoder;
+ }
public TypeEncoding<T> getEncoding(final T val)
{
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/ArrayType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/ArrayType.java
index 45b8dd5..32d6f85 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/ArrayType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/ArrayType.java
@@ -21,6 +21,7 @@
package org.apache.qpid.proton.codec;
import java.lang.reflect.Array;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
@@ -631,8 +632,13 @@
return decodeArrayAsObject(decoder, count);
}
-
-
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = decoder.readRawInt();
+ buffer.position(buffer.position() + size);
+ }
}
@@ -891,6 +897,13 @@
return decodeArrayAsObject(decoder, count);
}
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = ((int)decoder.readRawByte()) & 0xFF;
+ buffer.position(buffer.position() + size);
+ }
}
private BooleanType.BooleanEncoding getUnderlyingEncoding(final boolean[] a)
@@ -969,7 +982,7 @@
private static Object[] decodeArray(final DecoderImpl decoder, final int count)
{
- TypeConstructor constructor = decoder.readConstructor();
+ TypeConstructor constructor = decoder.readConstructor(true);
return decodeNonPrimitive(decoder, constructor, count);
}
@@ -1009,7 +1022,7 @@
private static Object decodeArrayAsObject(final DecoderImpl decoder, final int count)
{
- TypeConstructor constructor = decoder.readConstructor();
+ TypeConstructor constructor = decoder.readConstructor(true);
if(constructor.encodesJavaPrimitive())
{
if (count > decoder.getByteBufferRemaining()) {
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/BinaryType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/BinaryType.java
index 88c204f..1d739b8 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/BinaryType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/BinaryType.java
@@ -22,6 +22,7 @@
import org.apache.qpid.proton.amqp.Binary;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
@@ -53,7 +54,6 @@
return val.getLength() <= 255 ? _shortBinaryEncoding : _binaryEncoding;
}
-
public BinaryEncoding getCanonicalEncoding()
{
return _binaryEncoding;
@@ -64,6 +64,22 @@
return Arrays.asList(_shortBinaryEncoding, _binaryEncoding);
}
+ public void fastWrite(EncoderImpl encoder, Binary binary)
+ {
+ if (binary.getLength() <= 255)
+ {
+ encoder.writeRaw(EncodingCodes.VBIN8);
+ encoder.writeRaw((byte) binary.getLength());
+ encoder.writeRaw(binary.getArray(), binary.getArrayOffset(), binary.getLength());
+ }
+ else
+ {
+ encoder.writeRaw(EncodingCodes.VBIN32);
+ encoder.writeRaw(binary.getLength());
+ encoder.writeRaw(binary.getArray(), binary.getArrayOffset(), binary.getLength());
+ }
+ }
+
private class LongBinaryEncoding
extends LargeFloatingSizePrimitiveTypeEncoding<Binary>
implements BinaryEncoding
@@ -115,6 +131,14 @@
decoder.readRaw(data, 0, size);
return new Binary(data);
}
+
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = decoder.readRawInt();
+ buffer.position(buffer.position() + size);
+ }
}
private class ShortBinaryEncoding
@@ -163,5 +187,12 @@
getDecoder().readRaw(data, 0, size);
return new Binary(data);
}
+
+ public void skipValue()
+ {
+ ByteBuffer buffer = getDecoder().getByteBuffer();
+ int size = ((int)getDecoder().readRawByte()) & 0xff;
+ buffer.position(buffer.position() + size);
+ }
}
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/Decoder.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/Decoder.java
index 6053479..b1f47e0 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/Decoder.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/Decoder.java
@@ -145,5 +145,6 @@
void register(final Object descriptor, final DescribedTypeConstructor dtc);
+ void register(final Object descriptor, final FastPathDescribedTypeConstructor<?> dtc);
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/DecoderImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/DecoderImpl.java
index dd68f6a..1763129 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/DecoderImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/DecoderImpl.java
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.proton.codec;
+import org.apache.qpid.proton.ProtonException;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Decimal128;
import org.apache.qpid.proton.amqp.Decimal32;
@@ -31,48 +32,97 @@
import org.apache.qpid.proton.amqp.UnsignedLong;
import org.apache.qpid.proton.amqp.UnsignedShort;
+import java.io.IOException;
import java.lang.reflect.Array;
import java.nio.ByteBuffer;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.StandardCharsets;
import java.util.*;
public class DecoderImpl implements ByteBufferDecoder
{
-
private ByteBuffer _buffer;
- private PrimitiveTypeEncoding[] _constructors = new PrimitiveTypeEncoding[256];
- private Map<Object, DescribedTypeConstructor> _dynamicTypeConstructors =
+
+ private final CharsetDecoder _charsetDecoder = StandardCharsets.UTF_8.newDecoder();
+
+ private final PrimitiveTypeEncoding[] _constructors = new PrimitiveTypeEncoding[256];
+ private final Map<Object, DescribedTypeConstructor> _dynamicTypeConstructors =
new HashMap<Object, DescribedTypeConstructor>();
+ private final Map<Object, FastPathDescribedTypeConstructor<?>> _fastPathTypeConstructors =
+ new HashMap<Object, FastPathDescribedTypeConstructor<?>>();
public DecoderImpl()
{
}
-
DecoderImpl(final ByteBuffer buffer)
{
_buffer = buffer;
}
- TypeConstructor readConstructor()
+ public TypeConstructor<?> peekConstructor()
+ {
+ _buffer.mark();
+ try
+ {
+ return readConstructor();
+ }
+ finally
+ {
+ _buffer.reset();
+ }
+ }
+
+ @SuppressWarnings("rawtypes")
+ public TypeConstructor readConstructor()
+ {
+ return readConstructor(false);
+ }
+
+ @SuppressWarnings("rawtypes")
+ public TypeConstructor readConstructor(boolean excludeFastPathConstructors)
{
int code = ((int)readRawByte()) & 0xff;
if(code == EncodingCodes.DESCRIBED_TYPE_INDICATOR)
{
- final Object descriptor = readObject();
- TypeConstructor nestedEncoding = readConstructor();
- DescribedTypeConstructor dtc = _dynamicTypeConstructors.get(descriptor);
+ final byte encoding = _buffer.get(_buffer.position());
+ final Object descriptor;
+
+ if (EncodingCodes.SMALLULONG == encoding || EncodingCodes.ULONG == encoding)
+ {
+ descriptor = readUnsignedLong();
+ }
+ else if (EncodingCodes.SYM8 == encoding || EncodingCodes.SYM32 == encoding)
+ {
+ descriptor = readSymbol();
+ }
+ else
+ {
+ descriptor = readObject();
+ }
+
+ if (!excludeFastPathConstructors)
+ {
+ TypeConstructor<?> fastPathTypeConstructor = _fastPathTypeConstructors.get(descriptor);
+ if (fastPathTypeConstructor != null)
+ {
+ return fastPathTypeConstructor;
+ }
+ }
+
+ TypeConstructor<?> nestedEncoding = readConstructor();
+ DescribedTypeConstructor<?> dtc = _dynamicTypeConstructors.get(descriptor);
if(dtc == null)
{
dtc = new DescribedTypeConstructor()
{
-
public DescribedType newInstance(final Object described)
{
return new UnknownDescribedType(descriptor, described);
}
- public Class getTypeClass()
+ public Class<?> getTypeClass()
{
return UnknownDescribedType.class;
}
@@ -87,8 +137,15 @@
}
}
+ public void register(final Object descriptor, final FastPathDescribedTypeConstructor<?> btc)
+ {
+ _fastPathTypeConstructors.put(descriptor, btc);
+ }
+
public void register(final Object descriptor, final DescribedTypeConstructor dtc)
{
+ // Allow external type constructors to replace the built-in instances.
+ _fastPathTypeConstructors.remove(descriptor);
_dynamicTypeConstructors.put(descriptor, dtc);
}
@@ -108,37 +165,39 @@
public Boolean readBoolean(final Boolean defaultVal)
{
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ byte encodingCode = _buffer.get();
+
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.BOOLEAN_TRUE:
+ return (Boolean) _constructors[EncodingCodes.BOOLEAN_TRUE & 0xff].readValue();
+ case EncodingCodes.BOOLEAN_FALSE:
+ return (Boolean) _constructors[EncodingCodes.BOOLEAN_FALSE & 0xff].readValue();
+ case EncodingCodes.BOOLEAN:
+ return (Boolean) _constructors[EncodingCodes.BOOLEAN & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected boolean type but found encoding: " + encodingCode);
}
- else if(val instanceof Boolean)
- {
- return (Boolean) val;
- }
- throw unexpectedType(val, Boolean.class);
}
public boolean readBoolean(final boolean defaultVal)
{
- TypeConstructor constructor = readConstructor();
- if(constructor instanceof BooleanType.BooleanEncoding)
+ byte encodingCode = _buffer.get();
+
+ switch (encodingCode)
{
- return ((BooleanType.BooleanEncoding)constructor).readPrimitiveValue();
- }
- else
- {
- Object val = constructor.readValue();
- if(val == null)
- {
+ case EncodingCodes.BOOLEAN_TRUE:
+ return (Boolean) _constructors[EncodingCodes.BOOLEAN_TRUE & 0xff].readValue();
+ case EncodingCodes.BOOLEAN_FALSE:
+ return (Boolean) _constructors[EncodingCodes.BOOLEAN_FALSE & 0xff].readValue();
+ case EncodingCodes.BOOLEAN:
+ return (Boolean) _constructors[EncodingCodes.BOOLEAN & 0xff].readValue();
+ case EncodingCodes.NULL:
return defaultVal;
- }
- else
- {
- throw unexpectedType(val, Boolean.class);
- }
+ default:
+ throw new DecodeException("Expected boolean type but found encoding: " + encodingCode);
}
}
@@ -149,17 +208,16 @@
public Byte readByte(final Byte defaultVal)
{
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
- {
- return defaultVal;
+ byte encodingCode = _buffer.get();
+
+ switch (encodingCode) {
+ case EncodingCodes.BYTE:
+ return (Byte) _constructors[EncodingCodes.BYTE & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected byte type but found encoding: " + encodingCode);
}
- else if(val instanceof Byte)
- {
- return (Byte) val;
- }
- throw unexpectedType(val, Byte.class);
}
public byte readByte(final byte defaultVal)
@@ -190,18 +248,17 @@
public Short readShort(final Short defaultVal)
{
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
- {
- return defaultVal;
- }
- else if(val instanceof Short)
- {
- return (Short) val;
- }
- throw unexpectedType(val, Short.class);
+ byte encodingCode = _buffer.get();
+ switch (encodingCode)
+ {
+ case EncodingCodes.SHORT:
+ return (Short) _constructors[EncodingCodes.SHORT & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected Short type but found encoding: " + encodingCode);
+ }
}
public short readShort(final short defaultVal)
@@ -233,18 +290,19 @@
public Integer readInteger(final Integer defaultVal)
{
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
- {
- return defaultVal;
- }
- else if(val instanceof Integer)
- {
- return (Integer) val;
- }
- throw unexpectedType(val, Integer.class);
+ byte encodingCode = _buffer.get();
+ switch (encodingCode)
+ {
+ case EncodingCodes.SMALLINT:
+ return (Integer) _constructors[EncodingCodes.SMALLINT & 0xff].readValue();
+ case EncodingCodes.INT:
+ return (Integer) _constructors[EncodingCodes.INT & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected Integer type but found encoding: " + encodingCode);
+ }
}
public int readInteger(final int defaultVal)
@@ -276,19 +334,19 @@
public Long readLong(final Long defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.SMALLLONG:
+ return (Long) _constructors[EncodingCodes.SMALLLONG & 0xff].readValue();
+ case EncodingCodes.LONG:
+ return (Long) _constructors[EncodingCodes.LONG & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected Long type but found encoding: " + encodingCode);
}
- else if(val instanceof Long)
- {
- return (Long) val;
- }
- throw unexpectedType(val, Long.class);
-
}
public long readLong(final long defaultVal)
@@ -320,19 +378,17 @@
public UnsignedByte readUnsignedByte(final UnsignedByte defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.UBYTE:
+ return (UnsignedByte) _constructors[EncodingCodes.UBYTE & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected unsigned byte type but found encoding: " + encodingCode);
}
- else if(val instanceof UnsignedByte)
- {
- return (UnsignedByte) val;
- }
- throw unexpectedType(val, UnsignedByte.class);
-
}
public UnsignedShort readUnsignedShort()
@@ -342,19 +398,17 @@
public UnsignedShort readUnsignedShort(final UnsignedShort defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.USHORT:
+ return (UnsignedShort) _constructors[EncodingCodes.USHORT & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected UnsignedShort type but found encoding: " + encodingCode);
}
- else if(val instanceof UnsignedShort)
- {
- return (UnsignedShort) val;
- }
- throw unexpectedType(val, UnsignedShort.class);
-
}
public UnsignedInteger readUnsignedInteger()
@@ -364,19 +418,21 @@
public UnsignedInteger readUnsignedInteger(final UnsignedInteger defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.UINT0:
+ return (UnsignedInteger) _constructors[EncodingCodes.UINT0 & 0xff].readValue();
+ case EncodingCodes.SMALLUINT:
+ return (UnsignedInteger) _constructors[EncodingCodes.SMALLUINT & 0xff].readValue();
+ case EncodingCodes.UINT:
+ return (UnsignedInteger) _constructors[EncodingCodes.UINT & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected UnsignedInteger type but found encoding: " + encodingCode);
}
- else if(val instanceof UnsignedInteger)
- {
- return (UnsignedInteger) val;
- }
- throw unexpectedType(val, UnsignedInteger.class);
-
}
public UnsignedLong readUnsignedLong()
@@ -386,19 +442,21 @@
public UnsignedLong readUnsignedLong(final UnsignedLong defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.ULONG0:
+ return (UnsignedLong) _constructors[EncodingCodes.ULONG0 & 0xff].readValue();
+ case EncodingCodes.SMALLULONG:
+ return (UnsignedLong) _constructors[EncodingCodes.SMALLULONG & 0xff].readValue();
+ case EncodingCodes.ULONG:
+ return (UnsignedLong) _constructors[EncodingCodes.ULONG & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected UnsignedLong type but found encoding: " + encodingCode);
}
- else if(val instanceof UnsignedLong)
- {
- return (UnsignedLong) val;
- }
- throw unexpectedType(val, UnsignedLong.class);
-
}
public Character readCharacter()
@@ -408,40 +466,31 @@
public Character readCharacter(final Character defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.CHAR:
+ return (Character) _constructors[EncodingCodes.CHAR & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new DecodeException("Expected Character type but found encoding: " + encodingCode);
}
- else if(val instanceof Character)
- {
- return (Character) val;
- }
- throw unexpectedType(val, Character.class);
-
}
public char readCharacter(final char defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- if(constructor instanceof CharacterType.CharacterEncoding)
+ switch (encodingCode)
{
- return ((CharacterType.CharacterEncoding)constructor).readPrimitiveValue();
- }
- else
- {
- Object val = constructor.readValue();
- if(val == null)
- {
+ case EncodingCodes.CHAR:
+ return (Character) _constructors[EncodingCodes.CHAR & 0xff].readValue();
+ case EncodingCodes.NULL:
return defaultVal;
- }
- else
- {
- throw unexpectedType(val, Character.class);
- }
+ default:
+ throw new DecodeException("Expected Character type but found encoding: " + encodingCode);
}
}
@@ -452,19 +501,17 @@
public Float readFloat(final Float defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.FLOAT:
+ return (Float) _constructors[EncodingCodes.FLOAT & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new ProtonException("Expected Float type but found encoding: " + encodingCode);
}
- else if(val instanceof Float)
- {
- return (Float) val;
- }
- throw unexpectedType(val, Float.class);
-
}
public float readFloat(final float defaultVal)
@@ -496,19 +543,17 @@
public Double readDouble(final Double defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.DOUBLE:
+ return (Double) _constructors[EncodingCodes.DOUBLE & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new ProtonException("Expected Double type but found encoding: " + encodingCode);
}
- else if(val instanceof Double)
- {
- return (Double) val;
- }
- throw unexpectedType(val, Double.class);
-
}
public double readDouble(final double defaultVal)
@@ -540,19 +585,17 @@
public UUID readUUID(final UUID defaultVal)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultVal;
+ case EncodingCodes.UUID:
+ return (UUID) _constructors[EncodingCodes.UUID & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultVal;
+ default:
+ throw new ProtonException("Expected UUID type but found encoding: " + encodingCode);
}
- else if(val instanceof UUID)
- {
- return (UUID) val;
- }
- throw unexpectedType(val, UUID.class);
-
}
public Decimal32 readDecimal32()
@@ -562,19 +605,17 @@
public Decimal32 readDecimal32(final Decimal32 defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.DECIMAL32:
+ return (Decimal32) _constructors[EncodingCodes.DECIMAL32 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected Decimal32 type but found encoding: " + encodingCode);
}
- else if(val instanceof Decimal32)
- {
- return (Decimal32) val;
- }
- throw unexpectedType(val, Decimal32.class);
-
}
public Decimal64 readDecimal64()
@@ -584,18 +625,17 @@
public Decimal64 readDecimal64(final Decimal64 defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.DECIMAL64:
+ return (Decimal64) _constructors[EncodingCodes.DECIMAL64 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected Decimal64 type but found encoding: " + encodingCode);
}
- else if(val instanceof Decimal64)
- {
- return (Decimal64) val;
- }
- throw unexpectedType(val, Decimal64.class);
}
public Decimal128 readDecimal128()
@@ -605,18 +645,17 @@
public Decimal128 readDecimal128(final Decimal128 defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.DECIMAL128:
+ return (Decimal128) _constructors[EncodingCodes.DECIMAL128 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected Decimal128 type but found encoding: " + encodingCode);
}
- else if(val instanceof Decimal128)
- {
- return (Decimal128) val;
- }
- throw unexpectedType(val, Decimal128.class);
}
public Date readTimestamp()
@@ -626,18 +665,17 @@
public Date readTimestamp(final Date defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.TIMESTAMP:
+ return (Date) _constructors[EncodingCodes.TIMESTAMP & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected Timestamp type but found encoding: " + encodingCode);
}
- else if(val instanceof Date)
- {
- return (Date) val;
- }
- throw unexpectedType(val, Date.class);
}
public Binary readBinary()
@@ -647,18 +685,19 @@
public Binary readBinary(final Binary defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.VBIN8:
+ return (Binary) _constructors[EncodingCodes.VBIN8 & 0xff].readValue();
+ case EncodingCodes.VBIN32:
+ return (Binary) _constructors[EncodingCodes.VBIN32 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected Binary type but found encoding: " + encodingCode);
}
- else if(val instanceof Binary)
- {
- return (Binary) val;
- }
- throw unexpectedType(val, Binary.class);
}
public Symbol readSymbol()
@@ -668,18 +707,19 @@
public Symbol readSymbol(final Symbol defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.SYM8:
+ return (Symbol) _constructors[EncodingCodes.SYM8 & 0xff].readValue();
+ case EncodingCodes.SYM32:
+ return (Symbol) _constructors[EncodingCodes.SYM32 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected Symbol type but found encoding: " + encodingCode);
}
- else if(val instanceof Symbol)
- {
- return (Symbol) val;
- }
- throw unexpectedType(val, Symbol.class);
}
public String readString()
@@ -689,34 +729,39 @@
public String readString(final String defaultValue)
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return defaultValue;
+ case EncodingCodes.STR8:
+ return (String) _constructors[EncodingCodes.STR8 & 0xff].readValue();
+ case EncodingCodes.STR32:
+ return (String) _constructors[EncodingCodes.STR32 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return defaultValue;
+ default:
+ throw new ProtonException("Expected String type but found encoding: " + encodingCode);
}
- else if(val instanceof String)
- {
- return (String) val;
- }
- throw unexpectedType(val, String.class);
}
+ @SuppressWarnings("rawtypes")
public List readList()
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return null;
+ case EncodingCodes.LIST0:
+ return (List) _constructors[EncodingCodes.LIST0 & 0xff].readValue();
+ case EncodingCodes.LIST8:
+ return (List) _constructors[EncodingCodes.LIST8 & 0xff].readValue();
+ case EncodingCodes.LIST32:
+ return (List) _constructors[EncodingCodes.LIST32 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return null;
+ default:
+ throw new ProtonException("Expected List type but found encoding: " + encodingCode);
}
- else if(val instanceof List)
- {
- return (List) val;
- }
- throw unexpectedType(val, List.class);
}
public <T> void readList(final ListProcessor<T> processor)
@@ -724,20 +769,22 @@
//TODO.
}
+ @SuppressWarnings("rawtypes")
public Map readMap()
{
+ byte encodingCode = _buffer.get();
- TypeConstructor constructor = readConstructor();
- Object val = constructor.readValue();
- if(val == null)
+ switch (encodingCode)
{
- return null;
+ case EncodingCodes.MAP8:
+ return (Map) _constructors[EncodingCodes.MAP8 & 0xff].readValue();
+ case EncodingCodes.MAP32:
+ return (Map) _constructors[EncodingCodes.MAP32 & 0xff].readValue();
+ case EncodingCodes.NULL:
+ return null;
+ default:
+ throw new ProtonException("Expected Map type but found encoding: " + encodingCode);
}
- else if(val instanceof Map)
- {
- return (Map) val;
- }
- throw unexpectedType(val, Map.class);
}
public <T> T[] readArray(final Class<T> clazz)
@@ -877,14 +924,26 @@
public Object readObject()
{
+ boolean arrayType = false;
+ byte code = _buffer.get(_buffer.position());
+ switch (code)
+ {
+ case EncodingCodes.ARRAY8:
+ case EncodingCodes.ARRAY32:
+ arrayType = true;
+ }
+
TypeConstructor constructor = readConstructor();
if(constructor== null)
{
throw new DecodeException("Unknown constructor");
}
- return constructor instanceof ArrayType.ArrayEncoding
- ? ((ArrayType.ArrayEncoding)constructor).readValueArray()
- : constructor.readValue();
+
+ if (arrayType) {
+ return ((ArrayType.ArrayEncoding)constructor).readValueArray();
+ } else {
+ return constructor.readValue();
+ }
}
public Object readObject(final Object defaultValue)
@@ -942,7 +1001,7 @@
<V> V readRaw(TypeDecoder<V> decoder, int size)
{
- V decode = decoder.decode((ByteBuffer) _buffer.slice().limit(size));
+ V decode = decoder.decode(this, (ByteBuffer) _buffer.slice().limit(size));
_buffer.position(_buffer.position()+size);
return decode;
}
@@ -952,9 +1011,19 @@
_buffer = buffer;
}
+ public ByteBuffer getByteBuffer()
+ {
+ return _buffer;
+ }
+
+ CharsetDecoder getCharsetDecoder()
+ {
+ return _charsetDecoder;
+ }
+
interface TypeDecoder<V>
{
- V decode(ByteBuffer buf);
+ V decode(DecoderImpl decoder, ByteBuffer buf);
}
private static class UnknownDescribedType implements DescribedType
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/DynamicTypeConstructor.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/DynamicTypeConstructor.java
index 0cee927..2f8b9f9 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/DynamicTypeConstructor.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/DynamicTypeConstructor.java
@@ -53,6 +53,11 @@
return false;
}
+ public void skipValue()
+ {
+ _underlyingEncoding.skipValue();
+ }
+
public Class getTypeClass()
{
return _describedTypeConstructor.getTypeClass();
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/EncoderImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/EncoderImpl.java
index 97e1005..7bef1ae 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/EncoderImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/EncoderImpl.java
@@ -42,9 +42,9 @@
{
private static final byte DESCRIBED_TYPE_OP = (byte)0;
-
private WritableBuffer _buffer;
+ private final DecoderImpl _decoder;
private final Map<Class, AMQPType> _typeRegistry = new HashMap<Class, AMQPType>();
private Map<Object, AMQPType> _describedDescriptorRegistry = new HashMap<Object, AMQPType>();
private Map<Class, AMQPType> _describedTypesClassRegistry = new HashMap<Class, AMQPType>();
@@ -88,7 +88,7 @@
public EncoderImpl(DecoderImpl decoder)
{
-
+ _decoder = decoder;
_nullType = new NullType(this, decoder);
_booleanType = new BooleanType(this, decoder);
_byteType = new ByteType(this, decoder);
@@ -143,36 +143,32 @@
_buffer = buf;
}
+ public WritableBuffer getBuffer()
+ {
+ return _buffer;
+ }
+
+ public DecoderImpl getDecoder()
+ {
+ return _decoder;
+ }
@Override
public AMQPType getType(final Object element)
{
- if(element instanceof DescribedType)
- {
- AMQPType amqpType;
-
- Object descriptor = ((DescribedType)element).getDescriptor();
- amqpType = _describedDescriptorRegistry.get(descriptor);
- if(amqpType == null)
- {
- amqpType = new DynamicDescribedType(this, descriptor);
- _describedDescriptorRegistry.put(descriptor, amqpType);
- }
- return amqpType;
-
- }
- else
- {
- return getTypeFromClass(element == null ? Void.class : element.getClass());
- }
+ return getTypeFromClass(element == null ? Void.class : element.getClass(), element);
}
public AMQPType getTypeFromClass(final Class clazz)
{
+ return getTypeFromClass(clazz, null);
+ }
+
+ private AMQPType getTypeFromClass(final Class clazz, Object instance)
+ {
AMQPType amqpType = _typeRegistry.get(clazz);
if(amqpType == null)
{
-
if(clazz.isArray())
{
amqpType = _arrayType;
@@ -190,10 +186,21 @@
else if(DescribedType.class.isAssignableFrom(clazz))
{
amqpType = _describedTypesClassRegistry.get(clazz);
+ if(amqpType == null && instance != null)
+ {
+ Object descriptor = ((DescribedType) instance).getDescriptor();
+ amqpType = _describedDescriptorRegistry.get(descriptor);
+ if(amqpType == null)
+ {
+ amqpType = new DynamicDescribedType(this, descriptor);
+ _describedDescriptorRegistry.put(descriptor, amqpType);
+ }
+ }
}
}
_typeRegistry.put(clazz, amqpType);
}
+
return amqpType;
}
@@ -221,12 +228,19 @@
public void writeNull()
{
- _nullType.write();
+ _buffer.put(EncodingCodes.NULL);
}
public void writeBoolean(final boolean bool)
{
- _booleanType.writeValue(bool);
+ if (bool)
+ {
+ _buffer.put(EncodingCodes.BOOLEAN_TRUE);
+ }
+ else
+ {
+ _buffer.put(EncodingCodes.BOOLEAN_FALSE);
+ }
}
public void writeBoolean(final Boolean bool)
@@ -235,9 +249,13 @@
{
writeNull();
}
+ else if (Boolean.TRUE.equals(bool))
+ {
+ _buffer.put(EncodingCodes.BOOLEAN_TRUE);
+ }
else
{
- _booleanType.write(bool);
+ _buffer.put(EncodingCodes.BOOLEAN_FALSE);
}
}
@@ -249,7 +267,7 @@
}
else
{
- _unsignedByteType.write(ubyte);
+ _unsignedByteType.fastWrite(this, ubyte);
}
}
@@ -261,7 +279,7 @@
}
else
{
- _unsignedShortType.write(ushort);
+ _unsignedShortType.fastWrite(this, ushort);
}
}
@@ -273,7 +291,7 @@
}
else
{
- _unsignedIntegerType.write(uint);
+ _unsignedIntegerType.fastWrite(this, uint);
}
}
@@ -285,7 +303,7 @@
}
else
{
- _unsignedLongType.write(ulong);
+ _unsignedLongType.fastWrite(this, ulong);
}
}
@@ -446,9 +464,9 @@
}
}
- public void writeTimestamp(final long d)
+ public void writeTimestamp(final long timestamp)
{
- _timestampType.write(d);
+ _timestampType.fastWrite(this, timestamp);
}
public void writeTimestamp(final Date d)
@@ -459,7 +477,7 @@
}
else
{
- writeTimestamp(d.getTime());
+ _timestampType.fastWrite(this, d.getTime());
}
}
@@ -471,7 +489,7 @@
}
else
{
- _uuidType.write(uuid);
+ _uuidType.fastWrite(this, uuid);
}
}
@@ -484,7 +502,7 @@
}
else
{
- _binaryType.write(b);
+ _binaryType.fastWrite(this, b);
}
}
@@ -508,7 +526,7 @@
}
else
{
- _symbolType.write(s);
+ _symbolType.fastWrite(this, s);
}
}
@@ -662,56 +680,19 @@
public void writeObject(final Object o)
{
- AMQPType type = _typeRegistry.get(o == null ? Void.class : o.getClass());
+ if (o == null)
+ {
+ getBuffer().put(EncodingCodes.NULL);
+ return;
+ }
+
+ AMQPType type = _typeRegistry.get(o.getClass());
if(type == null)
{
if(o.getClass().isArray())
{
- Class<?> componentType = o.getClass().getComponentType();
- if(componentType.isPrimitive())
- {
- if(componentType == Boolean.TYPE)
- {
- writeArray((boolean[])o);
- }
- else if(componentType == Byte.TYPE)
- {
- writeArray((byte[])o);
- }
- else if(componentType == Short.TYPE)
- {
- writeArray((short[])o);
- }
- else if(componentType == Integer.TYPE)
- {
- writeArray((int[])o);
- }
- else if(componentType == Long.TYPE)
- {
- writeArray((long[])o);
- }
- else if(componentType == Float.TYPE)
- {
- writeArray((float[])o);
- }
- else if(componentType == Double.TYPE)
- {
- writeArray((double[])o);
- }
- else if(componentType == Character.TYPE)
- {
- writeArray((char[])o);
- }
- else
- {
- throw new IllegalArgumentException("Cannot write arrays of type " + componentType.getName());
- }
- }
- else
- {
- writeArray((Object[]) o);
- }
+ writeArrayType(o);
}
else if(o instanceof List)
{
@@ -727,9 +708,8 @@
}
else
{
- throw new IllegalArgumentException("Do not know how to write Objects of class " + o.getClass()
- .getName());
-
+ throw new IllegalArgumentException(
+ "Do not know how to write Objects of class " + o.getClass().getName());
}
}
else
@@ -738,6 +718,53 @@
}
}
+ private void writeArrayType(Object array) {
+ Class<?> componentType = array.getClass().getComponentType();
+ if(componentType.isPrimitive())
+ {
+ if(componentType == Boolean.TYPE)
+ {
+ writeArray((boolean[])array);
+ }
+ else if(componentType == Byte.TYPE)
+ {
+ writeArray((byte[])array);
+ }
+ else if(componentType == Short.TYPE)
+ {
+ writeArray((short[])array);
+ }
+ else if(componentType == Integer.TYPE)
+ {
+ writeArray((int[])array);
+ }
+ else if(componentType == Long.TYPE)
+ {
+ writeArray((long[])array);
+ }
+ else if(componentType == Float.TYPE)
+ {
+ writeArray((float[])array);
+ }
+ else if(componentType == Double.TYPE)
+ {
+ writeArray((double[])array);
+ }
+ else if(componentType == Character.TYPE)
+ {
+ writeArray((char[])array);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Cannot write arrays of type " + componentType.getName());
+ }
+ }
+ else
+ {
+ writeArray((Object[]) array);
+ }
+ }
+
public void writeRaw(final byte b)
{
_buffer.put(b);
@@ -814,4 +841,9 @@
}
}
}
+
+ AMQPType getNullTypeEncoder()
+ {
+ return _nullType;
+ }
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/FastPathDescribedTypeConstructor.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/FastPathDescribedTypeConstructor.java
new file mode 100644
index 0000000..13b3c80
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/FastPathDescribedTypeConstructor.java
@@ -0,0 +1,28 @@
+/*
+ * 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.proton.codec;
+
+/**
+ * Marker interface that indicates the TypeConstructor can decode known Proton-J types
+ * using a fast path read / write operation. These types may result in an encode that
+ * does not always write the smallest form of the given type to save time.
+ *
+ * @param <V> The type that this constructor handles
+ */
+public interface FastPathDescribedTypeConstructor<V> extends TypeConstructor<V> {
+
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/FixedSizePrimitiveTypeEncoding.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/FixedSizePrimitiveTypeEncoding.java
index 2762064..7c055ae 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/FixedSizePrimitiveTypeEncoding.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/FixedSizePrimitiveTypeEncoding.java
@@ -38,5 +38,10 @@
return getFixedSize();
}
+ public final void skipValue()
+ {
+ getDecoder().getByteBuffer().position(getDecoder().getByteBuffer().position() + getFixedSize());
+ }
+
protected abstract int getFixedSize();
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/ListType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/ListType.java
index 185373f..67e9d6e 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/ListType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/ListType.java
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.proton.codec;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -55,10 +56,9 @@
public ListEncoding getEncoding(final List val)
{
-
int calculatedSize = calculateSize(val, _encoder);
- ListEncoding encoding = val.isEmpty()
- ? _zeroListEncoding
+ ListEncoding encoding = val.isEmpty()
+ ? _zeroListEncoding
: (val.size() > 255 || calculatedSize >= 254)
? _listEncoding
: _shortListEncoding;
@@ -86,7 +86,6 @@
return len;
}
-
public ListEncoding getCanonicalEncoding()
{
return _listEncoding;
@@ -152,6 +151,8 @@
public List readValue()
{
DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+
int size = decoder.readRawInt();
// todo - limit the decoder with size
int count = decoder.readRawInt();
@@ -160,14 +161,78 @@
throw new IllegalArgumentException("List element count "+count+" is specified to be greater than the amount of data available ("+
decoder.getByteBufferRemaining()+")");
}
- List list = new ArrayList(count);
- for(int i = 0; i < count; i++)
+
+ TypeConstructor<?> typeConstructor = null;
+
+ List<Object> list = new ArrayList<>(count);
+ for (int i = 0; i < count; i++)
{
- list.add(decoder.readObject());
+ boolean arrayType = false;
+ byte code = buffer.get(buffer.position());
+ switch (code)
+ {
+ case EncodingCodes.ARRAY8:
+ case EncodingCodes.ARRAY32:
+ arrayType = true;
+ }
+
+ // Whenever we can just reuse the previously used TypeDecoder instead
+ // of spending time looking up the same one again.
+ if (typeConstructor == null)
+ {
+ typeConstructor = getDecoder().readConstructor();
+ }
+ else
+ {
+ buffer.mark();
+
+ byte encodingCode = buffer.get();
+ if (encodingCode == EncodingCodes.DESCRIBED_TYPE_INDICATOR || !(typeConstructor instanceof PrimitiveTypeEncoding<?>))
+ {
+ buffer.reset();
+ typeConstructor = getDecoder().readConstructor();
+ }
+ else
+ {
+ PrimitiveTypeEncoding<?> primitiveConstructor = (PrimitiveTypeEncoding<?>) typeConstructor;
+ if (encodingCode != primitiveConstructor.getEncodingCode())
+ {
+ buffer.reset();
+ typeConstructor = getDecoder().readConstructor();
+ }
+ }
+ }
+
+ if(typeConstructor == null)
+ {
+ throw new DecodeException("Unknown constructor");
+ }
+
+ final Object value;
+
+ if (arrayType)
+ {
+ value = ((ArrayType.ArrayEncoding) typeConstructor).readValueArray();
+ }
+ else
+ {
+ value = typeConstructor.readValue();
+ }
+
+ list.add(value);
}
+
return list;
}
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = decoder.readRawInt();
+ buffer.position(buffer.position() + size);
+ }
+
public void setValue(final List value, final int length)
{
_value = value;
@@ -229,19 +294,84 @@
public List readValue()
{
-
DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+
int size = ((int)decoder.readRawByte()) & 0xff;
// todo - limit the decoder with size
int count = ((int)decoder.readRawByte()) & 0xff;
- List list = new ArrayList(count);
- for(int i = 0; i < count; i++)
+
+ TypeConstructor<?> typeConstructor = null;
+
+ List<Object> list = new ArrayList<>(count);
+ for (int i = 0; i < count; i++)
{
- list.add(decoder.readObject());
+ boolean arrayType = false;
+ byte code = buffer.get(buffer.position());
+ switch (code)
+ {
+ case EncodingCodes.ARRAY8:
+ case EncodingCodes.ARRAY32:
+ arrayType = true;
+ }
+
+ // Whenever we can just reuse the previously used TypeDecoder instead
+ // of spending time looking up the same one again.
+ if (typeConstructor == null)
+ {
+ typeConstructor = getDecoder().readConstructor();
+ }
+ else
+ {
+ buffer.mark();
+
+ byte encodingCode = buffer.get();
+ if (encodingCode == EncodingCodes.DESCRIBED_TYPE_INDICATOR || !(typeConstructor instanceof PrimitiveTypeEncoding<?>))
+ {
+ buffer.reset();
+ typeConstructor = getDecoder().readConstructor();
+ }
+ else
+ {
+ PrimitiveTypeEncoding<?> primitiveConstructor = (PrimitiveTypeEncoding<?>) typeConstructor;
+ if (encodingCode != primitiveConstructor.getEncodingCode())
+ {
+ buffer.reset();
+ typeConstructor = getDecoder().readConstructor();
+ }
+ }
+ }
+
+ if (typeConstructor == null)
+ {
+ throw new DecodeException("Unknown constructor");
+ }
+
+ final Object value;
+
+ if (arrayType)
+ {
+ value = ((ArrayType.ArrayEncoding) typeConstructor).readValueArray();
+ }
+ else
+ {
+ value = typeConstructor.readValue();
+ }
+
+ list.add(value);
}
+
return list;
}
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = ((int)decoder.readRawByte()) & 0xff;
+ buffer.position(buffer.position() + size);
+ }
+
public void setValue(final List value, final int length)
{
_value = value;
@@ -249,7 +379,7 @@
}
}
-
+
private class ZeroListEncoding
extends FixedSizePrimitiveTypeEncoding<List>
implements ListEncoding
@@ -271,7 +401,6 @@
return 0;
}
-
public ListType getType()
{
return ListType.this;
@@ -294,7 +423,5 @@
{
return Collections.EMPTY_LIST;
}
-
-
}
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/MapType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/MapType.java
index 5c8a7c7..eba075a 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/MapType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/MapType.java
@@ -20,7 +20,13 @@
*/
package org.apache.qpid.proton.codec;
-import java.util.*;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
public class MapType extends AbstractPrimitiveType<Map>
{
@@ -28,6 +34,8 @@
private final MapEncoding _shortMapEncoding;
private EncoderImpl _encoder;
+ private AMQPType fixedKeyType;
+
private static interface MapEncoding extends PrimitiveTypeEncoding<Map>
{
void setValue(Map value, int length);
@@ -42,15 +50,21 @@
decoder.register(this);
}
+ @Override
public Class<Map> getTypeClass()
{
return Map.class;
}
+ public void setKeyEncoding(AMQPType<?> keyType)
+ {
+ this.fixedKeyType = keyType;
+ }
+
+ @Override
public MapEncoding getEncoding(final Map val)
{
-
- int calculatedSize = calculateSize(val, _encoder);
+ int calculatedSize = calculateSize(val, _encoder, fixedKeyType);
MapEncoding encoding = (val.size() > 127 || calculatedSize >= 254)
? _mapEncoding
: _shortMapEncoding;
@@ -59,29 +73,78 @@
return encoding;
}
- private static int calculateSize(final Map val, EncoderImpl encoder)
+ private static int calculateSize(final Map map, EncoderImpl encoder, AMQPType<?> fixedKeyType)
{
int len = 0;
- Iterator<Map.Entry> iter = val.entrySet().iterator();
+ Iterator<Map.Entry> iter = map.entrySet().iterator();
- while(iter.hasNext())
+ while (iter.hasNext())
{
Map.Entry element = iter.next();
- TypeEncoding elementEncoding = encoder.getType(element.getKey()).getEncoding(element.getKey());
+
+ AMQPType keyType = fixedKeyType;
+ if (fixedKeyType == null)
+ {
+ keyType = encoder.getType(element.getKey());
+ }
+
+ TypeEncoding elementEncoding = keyType.getEncoding(element.getKey());
len += elementEncoding.getConstructorSize()+elementEncoding.getValueSize(element.getKey());
elementEncoding = encoder.getType(element.getValue()).getEncoding(element.getValue());
len += elementEncoding.getConstructorSize()+elementEncoding.getValueSize(element.getValue());
-
}
return len;
}
+ private AMQPType<?> getKeyEncoding(EncoderImpl encoder, Object key)
+ {
+ if (fixedKeyType != null)
+ {
+ return fixedKeyType;
+ }
+ else
+ {
+ return encoder.getType(key);
+ }
+ }
+ private static TypeConstructor<?> findNextDecoder(DecoderImpl decoder, ByteBuffer buffer, TypeConstructor<?> previousConstructor)
+ {
+ if (previousConstructor == null)
+ {
+ return decoder.readConstructor();
+ }
+ else
+ {
+ buffer.mark();
+
+ byte encodingCode = buffer.get();
+ if (encodingCode == EncodingCodes.DESCRIBED_TYPE_INDICATOR || !(previousConstructor instanceof PrimitiveTypeEncoding<?>))
+ {
+ buffer.reset();
+ return decoder.readConstructor();
+ }
+ else
+ {
+ PrimitiveTypeEncoding<?> primitiveConstructor = (PrimitiveTypeEncoding<?>) previousConstructor;
+ if (encodingCode != primitiveConstructor.getEncodingCode())
+ {
+ buffer.reset();
+ return decoder.readConstructor();
+ }
+ }
+ }
+
+ return previousConstructor;
+ }
+
+ @Override
public MapEncoding getCanonicalEncoding()
{
return _mapEncoding;
}
+ @Override
public Collection<MapEncoding> getAllEncodings()
{
return Arrays.asList(_shortMapEncoding, _mapEncoding);
@@ -101,17 +164,23 @@
}
@Override
- protected void writeEncodedValue(final Map val)
+ protected void writeEncodedValue(final Map map)
{
- getEncoder().writeRaw(2 * val.size());
-
+ getEncoder().writeRaw(2 * map.size());
- Iterator<Map.Entry> iter = val.entrySet().iterator();
+ Iterator<Map.Entry> iter = map.entrySet().iterator();
- while(iter.hasNext())
+ while (iter.hasNext())
{
Map.Entry element = iter.next();
- TypeEncoding elementEncoding = getEncoder().getType(element.getKey()).getEncoding(element.getKey());
+
+ AMQPType keyType = fixedKeyType;
+ if (keyType == null)
+ {
+ keyType = getEncoder().getType(element.getKey());
+ }
+
+ TypeEncoding elementEncoding = keyType.getEncoding(element.getKey());
elementEncoding.writeConstructor();
elementEncoding.writeValue(element.getKey());
elementEncoding = getEncoder().getType(element.getValue()).getEncoding(element.getValue());
@@ -123,30 +192,33 @@
@Override
protected int getEncodedValueSize(final Map val)
{
- return 4 + ((val == _value) ? _length : calculateSize(val, getEncoder()));
+ return 4 + ((val == _value) ? _length : calculateSize(val, getEncoder(), fixedKeyType));
}
-
@Override
public byte getEncodingCode()
{
return EncodingCodes.MAP32;
}
+ @Override
public MapType getType()
{
return MapType.this;
}
+ @Override
public boolean encodesSuperset(final TypeEncoding<Map> encoding)
{
return (getType() == encoding.getType());
}
+ @Override
public Map readValue()
{
-
DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+
int size = decoder.readRawInt();
// todo - limit the decoder with size
int count = decoder.readRawInt();
@@ -154,17 +226,62 @@
throw new IllegalArgumentException("Map element count "+count+" is specified to be greater than the amount of data available ("+
decoder.getByteBufferRemaining()+")");
}
- Map map = new LinkedHashMap(count);
- for(int i = 0; i < count; i++)
+
+ TypeConstructor<?> keyConstructor = null;
+ TypeConstructor<?> valueConstructor = null;
+
+ Map<Object, Object> map = new LinkedHashMap<>(count);
+ for(int i = 0; i < count / 2; i++)
{
- Object key = decoder.readObject();
- i++;
- Object value = decoder.readObject();
+ keyConstructor = findNextDecoder(decoder, buffer, keyConstructor);
+ if(keyConstructor == null)
+ {
+ throw new DecodeException("Unknown constructor");
+ }
+
+ Object key = keyConstructor.readValue();
+
+ boolean arrayType = false;
+ byte code = buffer.get(buffer.position());
+ switch (code)
+ {
+ case EncodingCodes.ARRAY8:
+ case EncodingCodes.ARRAY32:
+ arrayType = true;
+ }
+
+ valueConstructor = findNextDecoder(decoder, buffer, valueConstructor);
+ if (valueConstructor == null)
+ {
+ throw new DecodeException("Unknown constructor");
+ }
+
+ final Object value;
+
+ if (arrayType)
+ {
+ value = ((ArrayType.ArrayEncoding) valueConstructor).readValueArray();
+ }
+ else
+ {
+ value = valueConstructor.readValue();
+ }
+
map.put(key, value);
}
+
return map;
}
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = decoder.readRawInt();
+ buffer.position(buffer.position() + size);
+ }
+
+ @Override
public void setValue(final Map value, final int length)
{
_value = value;
@@ -186,17 +303,22 @@
}
@Override
- protected void writeEncodedValue(final Map val)
+ protected void writeEncodedValue(final Map map)
{
- getEncoder().writeRaw((byte)(2*val.size()));
-
+ getEncoder().writeRaw((byte)(2 * map.size()));
- Iterator<Map.Entry> iter = val.entrySet().iterator();
-
- while(iter.hasNext())
+ Iterator<Map.Entry> iter = map.entrySet().iterator();
+ while (iter.hasNext())
{
Map.Entry element = iter.next();
- TypeEncoding elementEncoding = getEncoder().getType(element.getKey()).getEncoding(element.getKey());
+
+ AMQPType keyType = fixedKeyType;
+ if (keyType == null)
+ {
+ keyType = getEncoder().getType(element.getKey());
+ }
+
+ TypeEncoding elementEncoding = keyType.getEncoding(element.getKey());
elementEncoding.writeConstructor();
elementEncoding.writeValue(element.getKey());
elementEncoding = getEncoder().getType(element.getValue()).getEncoding(element.getValue());
@@ -208,44 +330,92 @@
@Override
protected int getEncodedValueSize(final Map val)
{
- return 1 + ((val == _value) ? _length : calculateSize(val, getEncoder()));
+ return 1 + ((val == _value) ? _length : calculateSize(val, getEncoder(), fixedKeyType));
}
-
@Override
public byte getEncodingCode()
{
return EncodingCodes.MAP8;
}
+ @Override
public MapType getType()
{
return MapType.this;
}
+ @Override
public boolean encodesSuperset(final TypeEncoding<Map> encoder)
{
return encoder == this;
}
+ @Override
public Map readValue()
{
DecoderImpl decoder = getDecoder();
- int size = ((int)decoder.readRawByte()) & 0xff;
- // todo - limit the decoder with size
- int count = ((int)decoder.readRawByte()) & 0xff;
+ ByteBuffer buffer = decoder.getByteBuffer();
- Map map = new LinkedHashMap(count);
- for(int i = 0; i < count; i++)
+ int size = (decoder.readRawByte()) & 0xff;
+ // todo - limit the decoder with size
+ int count = (decoder.readRawByte()) & 0xff;
+
+ TypeConstructor<?> keyConstructor = null;
+ TypeConstructor<?> valueConstructor = null;
+
+ Map<Object, Object> map = new LinkedHashMap<>(count);
+ for(int i = 0; i < count / 2; i++)
{
- Object key = decoder.readObject();
- i++;
- Object value = decoder.readObject();
+ keyConstructor = findNextDecoder(decoder, buffer, keyConstructor);
+ if(keyConstructor == null)
+ {
+ throw new DecodeException("Unknown constructor");
+ }
+
+ Object key = keyConstructor.readValue();
+
+ boolean arrayType = false;
+ byte code = buffer.get(buffer.position());
+ switch (code)
+ {
+ case EncodingCodes.ARRAY8:
+ case EncodingCodes.ARRAY32:
+ arrayType = true;
+ }
+
+ valueConstructor = findNextDecoder(decoder, buffer, valueConstructor);
+ if(valueConstructor== null)
+ {
+ throw new DecodeException("Unknown constructor");
+ }
+
+ final Object value;
+
+ if (arrayType)
+ {
+ value = ((ArrayType.ArrayEncoding) valueConstructor).readValueArray();
+ }
+ else
+ {
+ value = valueConstructor.readValue();
+ }
+
map.put(key, value);
}
+
return map;
}
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = ((int)decoder.readRawByte()) & 0xff;
+ buffer.position(buffer.position() + size);
+ }
+
+ @Override
public void setValue(final Map value, final int length)
{
_value = value;
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/StringType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/StringType.java
index a035e94..91476bc 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/StringType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/StringType.java
@@ -21,35 +21,34 @@
package org.apache.qpid.proton.codec;
import java.nio.ByteBuffer;
-import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
-import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Arrays;
import java.util.Collection;
public class StringType extends AbstractPrimitiveType<String>
{
- private static final Charset Charset_UTF8 = Charset.forName("UTF-8");
private static final DecoderImpl.TypeDecoder<String> _stringCreator =
- new DecoderImpl.TypeDecoder<String>()
+ new DecoderImpl.TypeDecoder<String>()
+ {
+
+ public String decode(DecoderImpl decoder, final ByteBuffer buf)
{
-
- public String decode(final ByteBuffer buf)
+ CharsetDecoder charsetDecoder = decoder.getCharsetDecoder();
+ try
{
- CharsetDecoder charsetDecoder = Charset_UTF8.newDecoder();
- try
- {
- CharBuffer charBuf = charsetDecoder.decode(buf);
- return charBuf.toString();
- }
- catch (CharacterCodingException e)
- {
- throw new IllegalArgumentException("Cannot parse String");
- }
-
+ return decoder.getCharsetDecoder().decode(buf).toString();
}
- };
+ catch (CharacterCodingException e)
+ {
+ throw new IllegalArgumentException("Cannot parse String");
+ }
+ finally
+ {
+ charsetDecoder.reset();
+ }
+ }
+ };
public static interface StringEncoding extends PrimitiveTypeEncoding<String>
@@ -175,6 +174,13 @@
_length = length;
}
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = decoder.readRawInt();
+ buffer.position(buffer.position() + size);
+ }
}
private class ShortStringEncoding
@@ -233,6 +239,14 @@
_value = val;
_length = length;
}
+
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = ((int)decoder.readRawByte()) & 0xff;
+ buffer.position(buffer.position() + size);
+ }
}
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/SymbolType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/SymbolType.java
index 4fb2038..e333e6a 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/SymbolType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/SymbolType.java
@@ -39,10 +39,9 @@
private DecoderImpl.TypeDecoder<Symbol> _symbolCreator =
new DecoderImpl.TypeDecoder<Symbol>()
{
-
- public Symbol decode(final ByteBuffer buf)
+ @Override
+ public Symbol decode(DecoderImpl decoder, ByteBuffer buf)
{
-
Symbol symbol = _symbolCache.get(buf);
if(symbol == null)
{
@@ -76,12 +75,27 @@
return Symbol.class;
}
+ public void fastWrite(EncoderImpl encoder, Symbol symbol)
+ {
+ if (symbol.length() <= 255)
+ {
+ encoder.writeRaw(EncodingCodes.SYM8);
+ encoder.writeRaw((byte) symbol.length());
+ symbol.writeTo(encoder.getBuffer());
+ }
+ else
+ {
+ encoder.writeRaw(EncodingCodes.SYM32);
+ encoder.writeRaw(symbol.length());
+ symbol.writeTo(encoder.getBuffer());
+ }
+ }
+
public SymbolEncoding getEncoding(final Symbol val)
{
return val.length() <= 255 ? _shortSymbolEncoding : _symbolEncoding;
}
-
public SymbolEncoding getCanonicalEncoding()
{
return _symbolEncoding;
@@ -105,13 +119,7 @@
@Override
protected void writeEncodedValue(final Symbol val)
{
- final int length = val.length();
- final EncoderImpl encoder = getEncoder();
-
- for(int i = 0; i < length; i++)
- {
- encoder.writeRaw((byte)val.charAt(i));
- }
+ val.writeTo(getEncoder().getBuffer());
}
@Override
@@ -143,8 +151,16 @@
int size = decoder.readRawInt();
return decoder.readRaw(_symbolCreator, size);
}
+
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = decoder.readRawInt();
+ buffer.position(buffer.position() + size);
+ }
}
-
+
private class ShortSymbolEncoding
extends SmallFloatingSizePrimitiveTypeEncoding<Symbol>
implements SymbolEncoding
@@ -158,14 +174,7 @@
@Override
protected void writeEncodedValue(final Symbol val)
{
-
- final int length = val.length();
- final EncoderImpl encoder = getEncoder();
-
- for(int i = 0; i < length; i++)
- {
- encoder.writeRaw((byte)val.charAt(i));
- }
+ val.writeTo(getEncoder().getBuffer());
}
@Override
@@ -197,5 +206,13 @@
int size = ((int)decoder.readRawByte()) & 0xff;
return decoder.readRaw(_symbolCreator, size);
}
+
+ public void skipValue()
+ {
+ DecoderImpl decoder = getDecoder();
+ ByteBuffer buffer = decoder.getByteBuffer();
+ int size = ((int)decoder.readRawByte()) & 0xff;
+ buffer.position(buffer.position() + size);
+ }
}
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/TimestampType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/TimestampType.java
index 54baa25..c2efe08 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/TimestampType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/TimestampType.java
@@ -45,6 +45,11 @@
return _timestampEncoding;
}
+ public void fastWrite(EncoderImpl encoder, long timestamp)
+ {
+ encoder.writeRaw(EncodingCodes.TIMESTAMP);
+ encoder.writeRaw(timestamp);
+ }
public TimestampEncoding getCanonicalEncoding()
{
@@ -60,7 +65,7 @@
{
_timestampEncoding.write(l);
}
-
+
private class TimestampEncoding extends FixedSizePrimitiveTypeEncoding<Date>
{
@@ -90,12 +95,12 @@
{
getEncoder().writeRaw(val.getTime());
}
-
+
public void write(final long l)
{
writeConstructor();
getEncoder().writeRaw(l);
-
+
}
public boolean encodesSuperset(final TypeEncoding<Date> encoding)
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/TypeConstructor.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/TypeConstructor.java
index 7b3f3a0..a707209 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/TypeConstructor.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/TypeConstructor.java
@@ -24,6 +24,8 @@
{
V readValue();
+ void skipValue();
+
boolean encodesJavaPrimitive();
Class<V> getTypeClass();
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/UUIDType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/UUIDType.java
index 20b9002..99f63f4 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/UUIDType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/UUIDType.java
@@ -45,6 +45,12 @@
return _uuidEncoding;
}
+ public void fastWrite(EncoderImpl encoder, UUID value)
+ {
+ encoder.writeRaw(EncodingCodes.UUID);
+ encoder.writeRaw(value.getMostSignificantBits());
+ encoder.writeRaw(value.getLeastSignificantBits());
+ }
public UUIDEncoding getCanonicalEncoding()
{
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedByteType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedByteType.java
index 781a9de..184d8be 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedByteType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedByteType.java
@@ -46,6 +46,11 @@
return _unsignedByteEncoding;
}
+ public void fastWrite(EncoderImpl encoder, UnsignedByte value)
+ {
+ encoder.writeRaw(EncodingCodes.UBYTE);
+ encoder.writeRaw(value.byteValue());
+ }
public UnsignedByteEncoding getCanonicalEncoding()
{
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedIntegerType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedIntegerType.java
index 860e373..ce64484 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedIntegerType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedIntegerType.java
@@ -59,6 +59,24 @@
: (i >= 0 && i <= 255) ? _smallUnsignedIntegerEncoding : _unsignedIntegerEncoding;
}
+ public void fastWrite(EncoderImpl encoder, UnsignedInteger value)
+ {
+ int intValue = value.intValue();
+ if (intValue == 0)
+ {
+ encoder.writeRaw(EncodingCodes.UINT0);
+ }
+ else if (intValue > 0 && intValue <= 255)
+ {
+ encoder.writeRaw(EncodingCodes.SMALLUINT);
+ encoder.writeRaw((byte)intValue);
+ }
+ else
+ {
+ encoder.writeRaw(EncodingCodes.UINT);
+ encoder.writeRaw(intValue);
+ }
+ }
public UnsignedIntegerEncoding getCanonicalEncoding()
{
@@ -102,12 +120,12 @@
{
getEncoder().writeRaw(val.intValue());
}
-
+
public void write(final int i)
{
writeConstructor();
getEncoder().writeRaw(i);
-
+
}
public boolean encodesSuperset(final TypeEncoding<UnsignedInteger> encoding)
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedLongType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedLongType.java
index 4b7980e..715028a 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedLongType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedLongType.java
@@ -35,8 +35,8 @@
private UnsignedLongEncoding _unsignedLongEncoding;
private UnsignedLongEncoding _smallUnsignedLongEncoding;
private UnsignedLongEncoding _zeroUnsignedLongEncoding;
-
-
+
+
UnsignedLongType(final EncoderImpl encoder, final DecoderImpl decoder)
{
_unsignedLongEncoding = new AllUnsignedLongEncoding(encoder, decoder);
@@ -59,6 +59,24 @@
: (l >= 0 && l <= 255L) ? _smallUnsignedLongEncoding : _unsignedLongEncoding;
}
+ public void fastWrite(EncoderImpl encoder, UnsignedLong value)
+ {
+ long longValue = value.longValue();
+ if (longValue == 0)
+ {
+ encoder.writeRaw(EncodingCodes.ULONG0);
+ }
+ else if (longValue > 0 && longValue <= 255)
+ {
+ encoder.writeRaw(EncodingCodes.SMALLULONG);
+ encoder.writeRaw((byte)longValue);
+ }
+ else
+ {
+ encoder.writeRaw(EncodingCodes.ULONG);
+ encoder.writeRaw(longValue);
+ }
+ }
public UnsignedLongEncoding getCanonicalEncoding()
{
@@ -157,8 +175,8 @@
return UnsignedLong.valueOf(((long)getDecoder().readRawByte())&0xffl);
}
}
-
-
+
+
private class ZeroUnsignedLongEncoding
extends FixedSizePrimitiveTypeEncoding<UnsignedLong>
implements UnsignedLongEncoding
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedShortType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedShortType.java
index 378c207..a30ee46 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedShortType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/UnsignedShortType.java
@@ -46,6 +46,11 @@
return _unsignedShortEncoder;
}
+ public void fastWrite(EncoderImpl encoder, UnsignedShort value)
+ {
+ encoder.writeRaw(EncodingCodes.USHORT);
+ encoder.writeRaw(value.shortValue());
+ }
public UnsignedShortEncoding getCanonicalEncoding()
{
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AcceptedType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AcceptedType.java
index 2da2d2b..32db983 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AcceptedType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AcceptedType.java
@@ -41,7 +41,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000024L);
- private AcceptedType(EncoderImpl encoder)
+ AcceptedType(EncoderImpl encoder)
{
super(encoder);
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpSequenceType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpSequenceType.java
index 8844893..054d34d 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpSequenceType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpSequenceType.java
@@ -42,7 +42,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000076L);
- private AmqpSequenceType(EncoderImpl encoder)
+ AmqpSequenceType(EncoderImpl encoder)
{
super(encoder);
}
@@ -68,7 +68,7 @@
return AmqpSequence.class;
}
-
+
public static void register(Decoder decoder, EncoderImpl encoder)
{
@@ -80,4 +80,3 @@
encoder.register(type);
}
}
-
\ No newline at end of file
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpValueType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpValueType.java
index d2dc8e1..9af847a 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpValueType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/AmqpValueType.java
@@ -42,7 +42,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000077L);
- private AmqpValueType(EncoderImpl encoder)
+ AmqpValueType(EncoderImpl encoder)
{
super(encoder);
}
@@ -68,7 +68,7 @@
return AmqpValue.class;
}
-
+
public static void register(Decoder decoder, EncoderImpl encoder)
{
@@ -80,4 +80,3 @@
encoder.register(type);
}
}
-
\ No newline at end of file
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/DataType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/DataType.java
index d7435c2..9ab06c0 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/DataType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/DataType.java
@@ -43,7 +43,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000075L);
- private DataType(EncoderImpl encoder)
+ DataType(EncoderImpl encoder)
{
super(encoder);
}
@@ -69,7 +69,7 @@
return Data.class;
}
-
+
public static void register(Decoder decoder, EncoderImpl encoder)
{
@@ -81,4 +81,3 @@
encoder.register(type);
}
}
-
\ No newline at end of file
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAcceptedType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAcceptedType.java
new file mode 100644
index 0000000..3624836
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAcceptedType.java
@@ -0,0 +1,123 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.Accepted;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathAcceptedType implements AMQPType<Accepted>, FastPathDescribedTypeConstructor<Accepted> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000024L), Symbol.valueOf("amqp:accepted:list"),
+ };
+
+ private final AcceptedType acceptedType;
+
+ public FastPathAcceptedType(EncoderImpl encoder) {
+ this.acceptedType = new AcceptedType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return acceptedType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return acceptedType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Accepted> getTypeClass() {
+ return Accepted.class;
+ }
+
+ @Override
+ public TypeEncoding<Accepted> getEncoding(Accepted accepted) {
+ return acceptedType.getEncoding(accepted);
+ }
+
+ @Override
+ public TypeEncoding<Accepted> getCanonicalEncoding() {
+ return acceptedType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Accepted>> getAllEncodings() {
+ return acceptedType.getAllEncodings();
+ }
+
+ @Override
+ public Accepted readValue() {
+ DecoderImpl decoder = getDecoder();
+ byte typeCode = decoder.getByteBuffer().get();
+
+ switch (typeCode) {
+ case EncodingCodes.LIST0:
+ break;
+ case EncodingCodes.LIST8:
+ decoder.getByteBuffer().get();
+ decoder.getByteBuffer().get();
+ break;
+ case EncodingCodes.LIST32:
+ decoder.getByteBuffer().getInt();
+ decoder.getByteBuffer().getInt();
+ break;
+ default:
+ throw new DecodeException("Incorrect type found in Accepted type encoding: " + typeCode);
+ }
+
+ return Accepted.getInstance();
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(Accepted accepted) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(acceptedType.getDescriptor());
+ buffer.put(EncodingCodes.LIST0);
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathAcceptedType type = new FastPathAcceptedType(encoder);
+ for(Object descriptor : DESCRIPTORS) {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAmqpSequenceType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAmqpSequenceType.java
new file mode 100644
index 0000000..611ec10
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAmqpSequenceType.java
@@ -0,0 +1,104 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathAmqpSequenceType implements AMQPType<AmqpSequence>, FastPathDescribedTypeConstructor<AmqpSequence> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000076L), Symbol.valueOf("amqp:amqp-sequence:list"),
+ };
+
+ private final AmqpSequenceType sequenceType;
+
+ public FastPathAmqpSequenceType(EncoderImpl encoder) {
+ this.sequenceType = new AmqpSequenceType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return sequenceType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return sequenceType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<AmqpSequence> getTypeClass() {
+ return AmqpSequence.class;
+ }
+
+ @Override
+ public TypeEncoding<AmqpSequence> getEncoding(AmqpSequence val) {
+ return sequenceType.getEncoding(val);
+ }
+
+ @Override
+ public TypeEncoding<AmqpSequence> getCanonicalEncoding() {
+ return sequenceType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<AmqpSequence>> getAllEncodings() {
+ return sequenceType.getAllEncodings();
+ }
+
+ @Override
+ public AmqpSequence readValue() {
+ return new AmqpSequence(getDecoder().readList());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(AmqpSequence sequence) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(sequenceType.getDescriptor());
+ getEncoder().writeObject(sequence.getValue());
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathAmqpSequenceType type = new FastPathAmqpSequenceType(encoder);
+ for (Object descriptor : DESCRIPTORS) {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAmqpValueType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAmqpValueType.java
new file mode 100644
index 0000000..7120481
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathAmqpValueType.java
@@ -0,0 +1,104 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathAmqpValueType implements AMQPType<AmqpValue>, FastPathDescribedTypeConstructor<AmqpValue> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000077L), Symbol.valueOf("amqp:amqp-value:*"),
+ };
+
+ private final AmqpValueType valueType;
+
+ public FastPathAmqpValueType(EncoderImpl encoder) {
+ this.valueType = new AmqpValueType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return valueType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return valueType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<AmqpValue> getTypeClass() {
+ return AmqpValue.class;
+ }
+
+ @Override
+ public TypeEncoding<AmqpValue> getEncoding(AmqpValue value) {
+ return valueType.getEncoding(value);
+ }
+
+ @Override
+ public TypeEncoding<AmqpValue> getCanonicalEncoding() {
+ return valueType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<AmqpValue>> getAllEncodings() {
+ return valueType.getAllEncodings();
+ }
+
+ @Override
+ public AmqpValue readValue() {
+ return new AmqpValue(getDecoder().readObject());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(AmqpValue value) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(valueType.getDescriptor());
+ getEncoder().writeObject(value.getValue());
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathAmqpValueType type = new FastPathAmqpValueType(encoder);
+ for(Object descriptor : DESCRIPTORS) {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathApplicationPropertiesType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathApplicationPropertiesType.java
new file mode 100644
index 0000000..4eed5e7
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathApplicationPropertiesType.java
@@ -0,0 +1,112 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.MapType;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathApplicationPropertiesType implements AMQPType<ApplicationProperties>, FastPathDescribedTypeConstructor<ApplicationProperties> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000074L), Symbol.valueOf("amqp:application-properties:map"),
+ };
+
+ private final ApplicationPropertiesType propertiesType;
+
+ public FastPathApplicationPropertiesType(EncoderImpl encoder) {
+ this.propertiesType = new ApplicationPropertiesType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return propertiesType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return propertiesType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<ApplicationProperties> getTypeClass() {
+ return ApplicationProperties.class;
+ }
+
+ @Override
+ public TypeEncoding<ApplicationProperties> getEncoding(ApplicationProperties val) {
+ return propertiesType.getEncoding(val);
+ }
+
+ @Override
+ public TypeEncoding<ApplicationProperties> getCanonicalEncoding() {
+ return propertiesType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<ApplicationProperties>> getAllEncodings() {
+ return propertiesType.getAllEncodings();
+ }
+
+ @Override
+ public ApplicationProperties readValue() {
+ return new ApplicationProperties(getDecoder().readMap());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(ApplicationProperties val) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(propertiesType.getDescriptor());
+
+ MapType mapType = (MapType) getEncoder().getType(val.getValue());
+
+ mapType.setKeyEncoding(getEncoder().getTypeFromClass(String.class));
+ mapType.write(val.getValue());
+ mapType.setKeyEncoding(null);
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathApplicationPropertiesType type = new FastPathApplicationPropertiesType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathDataType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathDataType.java
new file mode 100644
index 0000000..75a7543
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathDataType.java
@@ -0,0 +1,104 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.Data;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathDataType implements AMQPType<Data>, FastPathDescribedTypeConstructor<Data> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000075L), Symbol.valueOf("amqp:data:binary"),
+ };
+
+ private final DataType dataType;
+
+ public FastPathDataType(EncoderImpl encoder) {
+ this.dataType = new DataType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return dataType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return dataType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Data> getTypeClass() {
+ return dataType.getTypeClass();
+ }
+
+ @Override
+ public TypeEncoding<Data> getEncoding(Data val) {
+ return dataType.getEncoding(val);
+ }
+
+ @Override
+ public TypeEncoding<Data> getCanonicalEncoding() {
+ return dataType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Data>> getAllEncodings() {
+ return dataType.getAllEncodings();
+ }
+
+ @Override
+ public Data readValue() {
+ return new Data(getDecoder().readBinary());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(Data data) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(dataType.getDescriptor());
+ getEncoder().writeBinary(data.getValue());
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathDataType type = new FastPathDataType(encoder);
+ for(Object descriptor : DESCRIPTORS) {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathDeliveryAnnotationsType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathDeliveryAnnotationsType.java
new file mode 100644
index 0000000..4252393
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathDeliveryAnnotationsType.java
@@ -0,0 +1,113 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.MapType;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathDeliveryAnnotationsType implements AMQPType<DeliveryAnnotations>, FastPathDescribedTypeConstructor<DeliveryAnnotations> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000071L), Symbol.valueOf("amqp:delivery-annotations:map"),
+ };
+
+ private final DeliveryAnnotationsType annotationsType;
+
+ public FastPathDeliveryAnnotationsType(EncoderImpl encoder) {
+ this.annotationsType = new DeliveryAnnotationsType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return annotationsType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return annotationsType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<DeliveryAnnotations> getTypeClass() {
+ return DeliveryAnnotations.class;
+ }
+
+ @Override
+ public TypeEncoding<DeliveryAnnotations> getEncoding(DeliveryAnnotations val) {
+ return annotationsType.getEncoding(val);
+ }
+
+ @Override
+ public TypeEncoding<DeliveryAnnotations> getCanonicalEncoding() {
+ return annotationsType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<DeliveryAnnotations>> getAllEncodings() {
+ return annotationsType.getAllEncodings();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public DeliveryAnnotations readValue() {
+ return new DeliveryAnnotations(getDecoder().readMap());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(DeliveryAnnotations val) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(annotationsType.getDescriptor());
+
+ MapType mapType = (MapType) getEncoder().getType(val.getValue());
+
+ mapType.setKeyEncoding(getEncoder().getTypeFromClass(Symbol.class));
+ mapType.write(val.getValue());
+ mapType.setKeyEncoding(null);
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathDeliveryAnnotationsType type = new FastPathDeliveryAnnotationsType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathFooterType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathFooterType.java
new file mode 100644
index 0000000..d1b0d40
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathFooterType.java
@@ -0,0 +1,110 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.Footer;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.MapType;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathFooterType implements AMQPType<Footer>, FastPathDescribedTypeConstructor<Footer> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000078L), Symbol.valueOf("amqp:footer:map"),
+ };
+
+ private final FooterType footerType;
+
+ public FastPathFooterType(EncoderImpl encoder) {
+ this.footerType = new FooterType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return footerType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return footerType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Footer> getTypeClass() {
+ return Footer.class;
+ }
+
+ @Override
+ public TypeEncoding<Footer> getEncoding(Footer val) {
+ return footerType.getEncoding(val);
+ }
+
+ @Override
+ public TypeEncoding<Footer> getCanonicalEncoding() {
+ return footerType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Footer>> getAllEncodings() {
+ return footerType.getAllEncodings();
+ }
+
+ @Override
+ public Footer readValue() {
+ return new Footer(getDecoder().readMap());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(Footer val) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(footerType.getDescriptor());
+
+ MapType mapType = (MapType) getEncoder().getType(val.getValue());
+
+ mapType.write(val.getValue());
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathFooterType type = new FastPathFooterType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathHeaderType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathHeaderType.java
new file mode 100644
index 0000000..11c5223
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathHeaderType.java
@@ -0,0 +1,244 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.Header;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathHeaderType implements AMQPType<Header>, FastPathDescribedTypeConstructor<Header> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000070L), Symbol.valueOf("amqp:header:list"),
+ };
+
+ private final HeaderType headerType;
+
+ public FastPathHeaderType(EncoderImpl encoder) {
+ this.headerType = new HeaderType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return headerType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return headerType.getDecoder();
+ }
+
+ @Override
+ public Header readValue() {
+ DecoderImpl decoder = getDecoder();
+ byte typeCode = decoder.getByteBuffer().get();
+
+ @SuppressWarnings("unused")
+ int size = 0;
+ int count = 0;
+
+ switch (typeCode) {
+ case EncodingCodes.LIST0:
+ break;
+ case EncodingCodes.LIST8:
+ size = ((int)decoder.getByteBuffer().get()) & 0xff;
+ count = ((int)decoder.getByteBuffer().get()) & 0xff;
+ break;
+ case EncodingCodes.LIST32:
+ size = decoder.getByteBuffer().getInt();
+ count = decoder.getByteBuffer().getInt();
+ break;
+ default:
+ throw new DecodeException("Incorrect type found in Transfer encoding: " + typeCode);
+ }
+
+ Header header = new Header();
+
+ for (int index = 0; index < count; ++index) {
+ switch (index) {
+ case 0:
+ header.setDurable(decoder.readBoolean());
+ break;
+ case 1:
+ header.setPriority(decoder.readUnsignedByte());
+ break;
+ case 2:
+ header.setTtl(decoder.readUnsignedInteger());
+ break;
+ case 3:
+ header.setFirstAcquirer(decoder.readBoolean());
+ break;
+ case 4:
+ header.setDeliveryCount(decoder.readUnsignedInteger());
+ break;
+ default:
+ throw new IllegalStateException("To many entries in Header encoding");
+ }
+ }
+
+ return header;
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Header> getTypeClass() {
+ return Header.class;
+ }
+
+ @Override
+ public TypeEncoding<Header> getEncoding(Header header) {
+ return headerType.getEncoding(header);
+ }
+
+ @Override
+ public TypeEncoding<Header> getCanonicalEncoding() {
+ return headerType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Header>> getAllEncodings() {
+ return headerType.getAllEncodings();
+ }
+
+ @Override
+ public void write(Header value) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ int count = getElementCount(value);
+ byte encodingCode = deduceEncodingCode(value, count);
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(headerType.getDescriptor());
+
+ // Optimized step, no other data to be written.
+ if (count == 0 || encodingCode == EncodingCodes.LIST0) {
+ buffer.put(EncodingCodes.LIST0);
+ return;
+ }
+
+ final int fieldWidth;
+
+ if (encodingCode == EncodingCodes.LIST8) {
+ fieldWidth = 1;
+ buffer.put(EncodingCodes.LIST8);
+ } else {
+ fieldWidth = 4;
+ buffer.put(EncodingCodes.LIST32);
+ }
+
+ int startIndex = buffer.position();
+
+ // Reserve space for the size and write the count of list elements.
+ if (fieldWidth == 1) {
+ buffer.put((byte) 0);
+ buffer.put((byte) count);
+ } else {
+ buffer.putInt(0);
+ buffer.putInt(count);
+ }
+
+ // Write the list elements and then compute total size written.
+ for (int i = 0; i < count; ++i) {
+ writeElement(value, i);
+ }
+
+ // Move back and write the size
+ int endIndex = buffer.position();
+ int writeSize = endIndex - startIndex - fieldWidth;
+
+ buffer.position(startIndex);
+ if (fieldWidth == 1) {
+ buffer.put((byte) writeSize);
+ } else {
+ buffer.putInt(writeSize);
+ }
+ buffer.position(endIndex);
+ }
+
+ private void writeElement(Header header, int index) {
+ switch (index) {
+ case 0:
+ getEncoder().writeBoolean(header.getDurable());
+ break;
+ case 1:
+ getEncoder().writeUnsignedByte(header.getPriority());
+ break;
+ case 2:
+ getEncoder().writeUnsignedInteger(header.getTtl());
+ break;
+ case 3:
+ getEncoder().writeBoolean(header.getFirstAcquirer());
+ break;
+ case 4:
+ getEncoder().writeUnsignedInteger(header.getDeliveryCount());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown Header value index: " + index);
+ }
+ }
+
+ private int getElementCount(Header header) {
+ if (header.getDeliveryCount() != null) {
+ return 5;
+ } else if (header.getFirstAcquirer() != null) {
+ return 4;
+ } else if (header.getTtl() != null) {
+ return 3;
+ } else if (header.getPriority() != null) {
+ return 2;
+ } else if (header.getDurable() != null) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ private byte deduceEncodingCode(Header value, int elementCount) {
+ if (elementCount == 0) {
+ return EncodingCodes.LIST0;
+ } else {
+ return EncodingCodes.LIST8;
+ }
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathHeaderType type = new FastPathHeaderType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathMessageAnnotationsType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathMessageAnnotationsType.java
new file mode 100644
index 0000000..6d32004
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathMessageAnnotationsType.java
@@ -0,0 +1,113 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.MapType;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathMessageAnnotationsType implements AMQPType<MessageAnnotations>, FastPathDescribedTypeConstructor<MessageAnnotations> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000072L), Symbol.valueOf("amqp:message-annotations:map"),
+ };
+
+ private final MessageAnnotationsType annotationsType;
+
+ public FastPathMessageAnnotationsType(EncoderImpl encoder) {
+ this.annotationsType = new MessageAnnotationsType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return annotationsType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return annotationsType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<MessageAnnotations> getTypeClass() {
+ return MessageAnnotations.class;
+ }
+
+ @Override
+ public TypeEncoding<MessageAnnotations> getEncoding(MessageAnnotations val) {
+ return annotationsType.getEncoding(val);
+ }
+
+ @Override
+ public TypeEncoding<MessageAnnotations> getCanonicalEncoding() {
+ return annotationsType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<MessageAnnotations>> getAllEncodings() {
+ return annotationsType.getAllEncodings();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public MessageAnnotations readValue() {
+ return new MessageAnnotations(getDecoder().readMap());
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(MessageAnnotations val) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(annotationsType.getDescriptor());
+
+ MapType mapType = (MapType) getEncoder().getType(val.getValue());
+
+ mapType.setKeyEncoding(getEncoder().getTypeFromClass(Symbol.class));
+ mapType.write(val.getValue());
+ mapType.setKeyEncoding(null);
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathMessageAnnotationsType type = new FastPathMessageAnnotationsType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathPropertiesType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathPropertiesType.java
new file mode 100644
index 0000000..d689747
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathPropertiesType.java
@@ -0,0 +1,308 @@
+/*
+ * 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.proton.codec.messaging;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.Properties;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathPropertiesType implements AMQPType<Properties>, FastPathDescribedTypeConstructor<Properties> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000073L), Symbol.valueOf("amqp:properties:list"),
+ };
+
+ private final PropertiesType propertiesType;
+
+ public FastPathPropertiesType(EncoderImpl encoder) {
+ this.propertiesType = new PropertiesType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return propertiesType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return propertiesType.getDecoder();
+ }
+
+ @Override
+ public Properties readValue() {
+ DecoderImpl decoder = getDecoder();
+ byte typeCode = decoder.getByteBuffer().get();
+
+ @SuppressWarnings("unused")
+ int size = 0;
+ int count = 0;
+
+ switch (typeCode) {
+ case EncodingCodes.LIST0:
+ break;
+ case EncodingCodes.LIST8:
+ size = ((int)decoder.getByteBuffer().get()) & 0xff;
+ count = ((int)decoder.getByteBuffer().get()) & 0xff;
+ break;
+ case EncodingCodes.LIST32:
+ size = decoder.getByteBuffer().getInt();
+ count = decoder.getByteBuffer().getInt();
+ break;
+ default:
+ throw new DecodeException("Incorrect type found in Transfer encoding: " + typeCode);
+ }
+
+ Properties properties = new Properties();
+
+ for (int index = 0; index < count; ++index) {
+ switch (index) {
+ case 0:
+ properties.setMessageId(decoder.readObject());
+ break;
+ case 1:
+ properties.setUserId(decoder.readBinary());
+ break;
+ case 2:
+ properties.setTo(decoder.readString());
+ break;
+ case 3:
+ properties.setSubject(decoder.readString());
+ break;
+ case 4:
+ properties.setReplyTo(decoder.readString());
+ break;
+ case 5:
+ properties.setCorrelationId(decoder.readObject());
+ break;
+ case 6:
+ properties.setContentType(decoder.readSymbol());
+ break;
+ case 7:
+ properties.setContentEncoding(decoder.readSymbol());
+ break;
+ case 8:
+ properties.setAbsoluteExpiryTime(decoder.readTimestamp());
+ break;
+ case 9:
+ properties.setCreationTime(decoder.readTimestamp());
+ break;
+ case 10:
+ properties.setGroupId(decoder.readString());
+ break;
+ case 11:
+ properties.setGroupSequence(decoder.readUnsignedInteger());
+ break;
+ case 12:
+ properties.setReplyToGroupId(decoder.readString());
+ break;
+ default:
+ throw new IllegalStateException("To many entries in Properties encoding");
+ }
+ }
+
+ return properties;
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Properties> getTypeClass() {
+ return Properties.class;
+ }
+
+ @Override
+ public TypeEncoding<Properties> getEncoding(Properties properties) {
+ return propertiesType.getEncoding(properties);
+ }
+
+ @Override
+ public TypeEncoding<Properties> getCanonicalEncoding() {
+ return propertiesType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Properties>> getAllEncodings() {
+ return propertiesType.getAllEncodings();
+ }
+
+ @Override
+ public void write(Properties value) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ int count = getElementCount(value);
+ byte encodingCode = deduceEncodingCode(value, count);
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(propertiesType.getDescriptor());
+
+ // Optimized step, no other data to be written.
+ if (count == 0 || encodingCode == EncodingCodes.LIST0) {
+ buffer.put(EncodingCodes.LIST0);
+ return;
+ }
+
+ final int fieldWidth;
+
+ if (encodingCode == EncodingCodes.LIST8) {
+ fieldWidth = 1;
+ buffer.put(EncodingCodes.LIST8);
+ } else {
+ fieldWidth = 4;
+ buffer.put(EncodingCodes.LIST32);
+ }
+
+ int startIndex = buffer.position();
+
+ // Reserve space for the size and write the count of list elements.
+ if (fieldWidth == 1) {
+ buffer.put((byte) 0);
+ buffer.put((byte) count);
+ } else {
+ buffer.putInt(0);
+ buffer.putInt(count);
+ }
+
+ // Write the list elements and then compute total size written.
+ for (int i = 0; i < count; ++i) {
+ writeElement(value, i);
+ }
+
+ // Move back and write the size
+ int endIndex = buffer.position();
+ int writeSize = endIndex - startIndex - fieldWidth;
+
+ buffer.position(startIndex);
+ if (fieldWidth == 1) {
+ buffer.put((byte) writeSize);
+ } else {
+ buffer.putInt(writeSize);
+ }
+ buffer.position(endIndex);
+ }
+
+ private byte deduceEncodingCode(Properties value, int elementCount) {
+ if (elementCount == 0) {
+ return EncodingCodes.LIST0;
+ } else {
+ return EncodingCodes.LIST32;
+ }
+ }
+
+ private void writeElement(Properties properties, int index) {
+ switch (index) {
+ case 0:
+ getEncoder().writeObject(properties.getMessageId());
+ break;
+ case 1:
+ getEncoder().writeBinary(properties.getUserId());
+ break;
+ case 2:
+ getEncoder().writeString(properties.getTo());
+ break;
+ case 3:
+ getEncoder().writeString(properties.getSubject());
+ break;
+ case 4:
+ getEncoder().writeString(properties.getReplyTo());
+ break;
+ case 5:
+ getEncoder().writeObject(properties.getCorrelationId());
+ break;
+ case 6:
+ getEncoder().writeSymbol(properties.getContentType());
+ break;
+ case 7:
+ getEncoder().writeSymbol(properties.getContentEncoding());
+ break;
+ case 8:
+ getEncoder().writeTimestamp(properties.getAbsoluteExpiryTime());
+ break;
+ case 9:
+ getEncoder().writeTimestamp(properties.getCreationTime());
+ break;
+ case 10:
+ getEncoder().writeString(properties.getGroupId());
+ break;
+ case 11:
+ getEncoder().writeUnsignedInteger(properties.getGroupSequence());
+ break;
+ case 12:
+ getEncoder().writeString(properties.getReplyToGroupId());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown Properties value index: " + index);
+ }
+ }
+
+ private int getElementCount(Properties properties) {
+ if (properties.getReplyToGroupId() != null) {
+ return 13;
+ } else if (properties.getGroupSequence() != null) {
+ return 12;
+ } else if (properties.getGroupId() != null) {
+ return 11;
+ } else if (properties.getCreationTime() != null) {
+ return 10;
+ } else if (properties.getAbsoluteExpiryTime() != null) {
+ return 9;
+ } else if (properties.getContentEncoding() != null) {
+ return 8;
+ } else if (properties.getContentType() != null) {
+ return 7;
+ } else if (properties.getCorrelationId() != null) {
+ return 6;
+ } else if (properties.getReplyTo() != null) {
+ return 5;
+ } else if (properties.getSubject() != null) {
+ return 4;
+ } else if (properties.getTo() != null) {
+ return 3;
+ } else if (properties.getUserId() != null) {
+ return 2;
+ } else if (properties.getMessageId() != null) {
+ return 1;
+ }
+
+ return 0;
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathPropertiesType type = new FastPathPropertiesType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/HeaderType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/HeaderType.java
index b7f5a43..310bce1 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/HeaderType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/HeaderType.java
@@ -35,7 +35,6 @@
import org.apache.qpid.proton.codec.DescribedTypeConstructor;
import org.apache.qpid.proton.codec.EncoderImpl;
-
public class HeaderType extends AbstractDescribedType<Header,List> implements DescribedTypeConstructor<Header>
{
private static final Object[] DESCRIPTORS =
@@ -154,5 +153,4 @@
}
encoder.register(type);
}
-
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/PropertiesType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/PropertiesType.java
index 818aaa8..98c4431 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/PropertiesType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/PropertiesType.java
@@ -36,7 +36,6 @@
import org.apache.qpid.proton.codec.DescribedTypeConstructor;
import org.apache.qpid.proton.codec.EncoderImpl;
-
public class PropertiesType extends AbstractDescribedType<Properties,List> implements DescribedTypeConstructor<Properties>
{
private static final Object[] DESCRIPTORS =
@@ -46,7 +45,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000073L);
- private PropertiesType(EncoderImpl encoder)
+ PropertiesType(EncoderImpl encoder)
{
super(encoder);
}
@@ -190,8 +189,6 @@
return Properties.class;
}
-
-
public static void register(Decoder decoder, EncoderImpl encoder)
{
PropertiesType type = new PropertiesType(encoder);
@@ -201,5 +198,4 @@
}
encoder.register(type);
}
-
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/DispositionType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/DispositionType.java
index b833613..44a6202 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/DispositionType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/DispositionType.java
@@ -47,7 +47,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000015L);
- private DispositionType(EncoderImpl encoder)
+ DispositionType(EncoderImpl encoder)
{
super(encoder);
}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathDispositionType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathDispositionType.java
new file mode 100644
index 0000000..01e18e7
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathDispositionType.java
@@ -0,0 +1,248 @@
+/*
+ * 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.proton.codec.transport;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.messaging.Accepted;
+import org.apache.qpid.proton.amqp.messaging.Released;
+import org.apache.qpid.proton.amqp.transport.DeliveryState;
+import org.apache.qpid.proton.amqp.transport.Disposition;
+import org.apache.qpid.proton.amqp.transport.Role;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathDispositionType implements AMQPType<Disposition>, FastPathDescribedTypeConstructor<Disposition> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000015L), Symbol.valueOf("amqp:disposition:list"),
+ };
+
+ private final DispositionType dispositionType;
+
+ public FastPathDispositionType(EncoderImpl encoder) {
+ this.dispositionType = new DispositionType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return dispositionType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return dispositionType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Disposition> getTypeClass() {
+ return Disposition.class;
+ }
+
+ @Override
+ public TypeEncoding<Disposition> getEncoding(Disposition disposition) {
+ return dispositionType.getEncoding(disposition);
+ }
+
+ @Override
+ public TypeEncoding<Disposition> getCanonicalEncoding() {
+ return dispositionType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Disposition>> getAllEncodings() {
+ return dispositionType.getAllEncodings();
+ }
+
+ @Override
+ public Disposition readValue() {
+ DecoderImpl decoder = getDecoder();
+ byte typeCode = decoder.getByteBuffer().get();
+
+ @SuppressWarnings("unused")
+ int size = 0;
+ int count = 0;
+
+ switch (typeCode) {
+ case EncodingCodes.LIST0:
+ // TODO - Technically invalid however old decoder also allowed this.
+ break;
+ case EncodingCodes.LIST8:
+ size = ((int)decoder.getByteBuffer().get()) & 0xff;
+ count = ((int)decoder.getByteBuffer().get()) & 0xff;
+ break;
+ case EncodingCodes.LIST32:
+ size = decoder.getByteBuffer().getInt();
+ count = decoder.getByteBuffer().getInt();
+ break;
+ default:
+ throw new DecodeException("Incorrect type found in Disposition encoding: " + typeCode);
+ }
+
+ Disposition disposition = new Disposition();
+
+ for (int index = 0; index < count; ++index) {
+ switch (index) {
+ case 0:
+ disposition.setRole(Boolean.TRUE.equals(decoder.readBoolean()) ? Role.RECEIVER : Role.SENDER);
+ break;
+ case 1:
+ disposition.setFirst(decoder.readUnsignedInteger());
+ break;
+ case 2:
+ disposition.setLast(decoder.readUnsignedInteger());
+ break;
+ case 3:
+ disposition.setSettled(decoder.readBoolean(false));
+ break;
+ case 4:
+ disposition.setState((DeliveryState) decoder.readObject());
+ break;
+ case 5:
+ disposition.setBatchable(decoder.readBoolean(false));
+ break;
+ default:
+ throw new IllegalStateException("To many entries in Disposition encoding");
+ }
+ }
+
+ return disposition;
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(Disposition disposition) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ int count = getElementCount(disposition);
+ byte encodingCode = deduceEncodingCode(disposition, count);
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(dispositionType.getDescriptor());
+
+ final int fieldWidth;
+
+ if (encodingCode == EncodingCodes.LIST8) {
+ fieldWidth = 1;
+ buffer.put(EncodingCodes.LIST8);
+ } else {
+ fieldWidth = 4;
+ buffer.put(EncodingCodes.LIST32);
+ }
+
+ int startIndex = buffer.position();
+
+ // Reserve space for the size and write the count of list elements.
+ if (fieldWidth == 1) {
+ buffer.put((byte) 0);
+ buffer.put((byte) count);
+ } else {
+ buffer.putInt(0);
+ buffer.putInt(count);
+ }
+
+ // Write the list elements and then compute total size written.
+ for (int i = 0; i < count; ++i) {
+ writeElement(disposition, i);
+ }
+
+ // Move back and write the size
+ int endIndex = buffer.position();
+ int writeSize = endIndex - startIndex - fieldWidth;
+
+ buffer.position(startIndex);
+ if (fieldWidth == 1) {
+ buffer.put((byte) writeSize);
+ } else {
+ buffer.putInt(writeSize);
+ }
+ buffer.position(endIndex);
+ }
+
+ private void writeElement(Disposition disposition, int index) {
+ switch (index) {
+ case 0:
+ getEncoder().writeBoolean(disposition.getRole().getValue());
+ break;
+ case 1:
+ getEncoder().writeUnsignedInteger(disposition.getFirst());
+ break;
+ case 2:
+ getEncoder().writeUnsignedInteger(disposition.getLast());
+ break;
+ case 3:
+ getEncoder().writeBoolean(disposition.getSettled());
+ break;
+ case 4:
+ getEncoder().writeObject(disposition.getState());
+ break;
+ case 5:
+ getEncoder().writeBoolean(disposition.getBatchable());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown Disposition value index: " + index);
+ }
+ }
+
+ private int getElementCount(Disposition disposition) {
+ if (disposition.getBatchable()) {
+ return 6;
+ } else if (disposition.getState() != null) {
+ return 5;
+ } else if (disposition.getSettled()) {
+ return 4;
+ } else if (disposition.getLast() != null) {
+ return 3;
+ } else {
+ return 2;
+ }
+ }
+
+ private byte deduceEncodingCode(Disposition value, int elementCount) {
+ if (value.getState() == null) {
+ return EncodingCodes.LIST8;
+ } else if (value.getState() == Accepted.getInstance() || value.getState() == Released.getInstance()) {
+ return EncodingCodes.LIST8;
+ } else {
+ return EncodingCodes.LIST32;
+ }
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathDispositionType type = new FastPathDispositionType(encoder);
+ for(Object descriptor : DESCRIPTORS) {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathFlowType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathFlowType.java
new file mode 100644
index 0000000..78abc5c
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathFlowType.java
@@ -0,0 +1,279 @@
+/*
+ * 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.proton.codec.transport;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.transport.Flow;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+public class FastPathFlowType implements AMQPType<Flow>, FastPathDescribedTypeConstructor<Flow> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000013L), Symbol.valueOf("amqp:flow:list"),
+ };
+
+ private final FlowType flowType;
+
+ public FastPathFlowType(EncoderImpl encoder) {
+ this.flowType = new FlowType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return flowType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return flowType.getDecoder();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Flow> getTypeClass() {
+ return Flow.class;
+ }
+
+ @Override
+ public TypeEncoding<Flow> getEncoding(Flow flow) {
+ return flowType.getEncoding(flow);
+ }
+
+ @Override
+ public TypeEncoding<Flow> getCanonicalEncoding() {
+ return flowType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Flow>> getAllEncodings() {
+ return flowType.getAllEncodings();
+ }
+
+ @Override
+ public Flow readValue() {
+ DecoderImpl decoder = getDecoder();
+ byte typeCode = decoder.getByteBuffer().get();
+
+ @SuppressWarnings("unused")
+ int size = 0;
+ int count = 0;
+
+ switch (typeCode) {
+ case EncodingCodes.LIST0:
+ // TODO - Technically invalid however old decoder also allowed this.
+ break;
+ case EncodingCodes.LIST8:
+ size = ((int)decoder.getByteBuffer().get()) & 0xff;
+ count = ((int)decoder.getByteBuffer().get()) & 0xff;
+ break;
+ case EncodingCodes.LIST32:
+ size = decoder.getByteBuffer().getInt();
+ count = decoder.getByteBuffer().getInt();
+ break;
+ default:
+ throw new DecodeException("Incorrect type found in Flow encoding: " + typeCode);
+ }
+
+ Flow flow = new Flow();
+
+ for (int index = 0; index < count; ++index) {
+ switch (index) {
+ case 0:
+ flow.setNextIncomingId(decoder.readUnsignedInteger());
+ break;
+ case 1:
+ flow.setIncomingWindow(decoder.readUnsignedInteger());
+ break;
+ case 2:
+ flow.setNextOutgoingId(decoder.readUnsignedInteger());
+ break;
+ case 3:
+ flow.setOutgoingWindow(decoder.readUnsignedInteger());
+ break;
+ case 4:
+ flow.setHandle(decoder.readUnsignedInteger());
+ break;
+ case 5:
+ flow.setDeliveryCount(decoder.readUnsignedInteger());
+ break;
+ case 6:
+ flow.setLinkCredit(decoder.readUnsignedInteger());
+ break;
+ case 7:
+ flow.setAvailable(decoder.readUnsignedInteger());
+ break;
+ case 8:
+ flow.setDrain(decoder.readBoolean(false));
+ break;
+ case 9:
+ flow.setEcho(decoder.readBoolean(false));
+ break;
+ case 10:
+ flow.setProperties(decoder.readMap());
+ break;
+ default:
+ throw new IllegalStateException("To many entries in Flow encoding");
+ }
+ }
+
+ return flow;
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public void write(Flow flow) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ int count = getElementCount(flow);
+ byte encodingCode = deduceEncodingCode(flow, count);
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(flowType.getDescriptor());
+
+ final int fieldWidth;
+
+ if (encodingCode == EncodingCodes.LIST8) {
+ fieldWidth = 1;
+ buffer.put(EncodingCodes.LIST8);
+ } else {
+ fieldWidth = 4;
+ buffer.put(EncodingCodes.LIST32);
+ }
+
+ int startIndex = buffer.position();
+
+ // Reserve space for the size and write the count of list elements.
+ if (fieldWidth == 1) {
+ buffer.put((byte) 0);
+ buffer.put((byte) count);
+ } else {
+ buffer.putInt(0);
+ buffer.putInt(count);
+ }
+
+ // Write the list elements and then compute total size written.
+ for (int i = 0; i < count; ++i) {
+ writeElement(flow, i);
+ }
+
+ // Move back and write the size
+ int endIndex = buffer.position();
+ int writeSize = endIndex - startIndex - fieldWidth;
+
+ buffer.position(startIndex);
+ if (fieldWidth == 1) {
+ buffer.put((byte) writeSize);
+ } else {
+ buffer.putInt(writeSize);
+ }
+ buffer.position(endIndex);
+ }
+
+ private void writeElement(Flow flow, int index) {
+ switch (index) {
+ case 0:
+ getEncoder().writeUnsignedInteger(flow.getNextIncomingId());
+ break;
+ case 1:
+ getEncoder().writeUnsignedInteger(flow.getIncomingWindow());
+ break;
+ case 2:
+ getEncoder().writeUnsignedInteger(flow.getNextOutgoingId());
+ break;
+ case 3:
+ getEncoder().writeUnsignedInteger(flow.getOutgoingWindow());
+ break;
+ case 4:
+ getEncoder().writeUnsignedInteger(flow.getHandle());
+ break;
+ case 5:
+ getEncoder().writeUnsignedInteger(flow.getDeliveryCount());
+ break;
+ case 6:
+ getEncoder().writeUnsignedInteger(flow.getLinkCredit());
+ break;
+ case 7:
+ getEncoder().writeUnsignedInteger(flow.getAvailable());
+ break;
+ case 8:
+ getEncoder().writeBoolean(flow.getDrain());
+ break;
+ case 9:
+ getEncoder().writeBoolean(flow.getEcho());
+ break;
+ case 10:
+ getEncoder().writeMap(flow.getProperties());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown Flow value index: " + index);
+ }
+ }
+
+ private int getElementCount(Flow flow) {
+ if (flow.getProperties() != null) {
+ return 11;
+ } else if (flow.getEcho()) {
+ return 10;
+ } else if (flow.getDrain()) {
+ return 9;
+ } else if (flow.getAvailable() != null) {
+ return 8;
+ } else if (flow.getLinkCredit() != null) {
+ return 7;
+ } else if (flow.getDeliveryCount() != null) {
+ return 6;
+ } else if (flow.getHandle() != null) {
+ return 5;
+ } else {
+ return 4;
+ }
+ }
+
+ private byte deduceEncodingCode(Flow value, int elementCount) {
+ if (value.getProperties() == null) {
+ return EncodingCodes.LIST8;
+ } else {
+ return EncodingCodes.LIST32;
+ }
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathFlowType type = new FastPathFlowType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathTransferType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathTransferType.java
new file mode 100644
index 0000000..685890a
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FastPathTransferType.java
@@ -0,0 +1,295 @@
+/*
+ * 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.proton.codec.transport;
+
+import java.util.Collection;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedByte;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+import org.apache.qpid.proton.amqp.transport.DeliveryState;
+import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
+import org.apache.qpid.proton.amqp.transport.Transfer;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.FastPathDescribedTypeConstructor;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.Decoder;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+
+/**
+ * Fast TrasnferType encoder
+ */
+public class FastPathTransferType implements AMQPType<Transfer>, FastPathDescribedTypeConstructor<Transfer> {
+
+ private static final Object[] DESCRIPTORS =
+ {
+ UnsignedLong.valueOf(0x0000000000000014L), Symbol.valueOf("amqp:transfer:list"),
+ };
+
+ private final TransferType transferType;
+
+ public FastPathTransferType(EncoderImpl encoder) {
+ this.transferType = new TransferType(encoder);
+ }
+
+ public EncoderImpl getEncoder() {
+ return transferType.getEncoder();
+ }
+
+ public DecoderImpl getDecoder() {
+ return transferType.getDecoder();
+ }
+
+ @Override
+ public Transfer readValue() {
+ DecoderImpl decoder = getDecoder();
+ byte typeCode = decoder.getByteBuffer().get();
+
+ @SuppressWarnings("unused")
+ int size = 0;
+ int count = 0;
+
+ switch (typeCode) {
+ case EncodingCodes.LIST0:
+ // TODO - Technically invalid however old decoder also allowed this.
+ break;
+ case EncodingCodes.LIST8:
+ size = ((int)decoder.getByteBuffer().get()) & 0xff;
+ count = ((int)decoder.getByteBuffer().get()) & 0xff;
+ break;
+ case EncodingCodes.LIST32:
+ size = decoder.getByteBuffer().getInt();
+ count = decoder.getByteBuffer().getInt();
+ break;
+ default:
+ throw new DecodeException("Incorrect type found in Transfer encoding: " + typeCode);
+ }
+
+ Transfer transfer = new Transfer();
+
+ for (int index = 0; index < count; ++index) {
+ switch (index) {
+ case 0:
+ transfer.setHandle(decoder.readUnsignedInteger());
+ break;
+ case 1:
+ transfer.setDeliveryId(decoder.readUnsignedInteger());
+ break;
+ case 2:
+ transfer.setDeliveryTag(decoder.readBinary());
+ break;
+ case 3:
+ transfer.setMessageFormat(decoder.readUnsignedInteger());
+ break;
+ case 4:
+ transfer.setSettled(decoder.readBoolean());
+ break;
+ case 5:
+ transfer.setMore(decoder.readBoolean(false));
+ break;
+ case 6:
+ UnsignedByte rcvSettleMode = decoder.readUnsignedByte();
+ transfer.setRcvSettleMode(rcvSettleMode == null ? null : ReceiverSettleMode.values()[rcvSettleMode.intValue()]);
+ break;
+ case 7:
+ transfer.setState((DeliveryState) decoder.readObject());
+ break;
+ case 8:
+ transfer.setResume(decoder.readBoolean(false));
+ break;
+ case 9:
+ transfer.setAborted(decoder.readBoolean(false));
+ break;
+ case 10:
+ transfer.setBatchable(decoder.readBoolean(false));
+ break;
+ default:
+ throw new IllegalStateException("To many entries in Transfer encoding");
+ }
+ }
+
+ return transfer;
+ }
+
+ @Override
+ public void skipValue() {
+ getDecoder().readConstructor().skipValue();
+ }
+
+ @Override
+ public boolean encodesJavaPrimitive() {
+ return false;
+ }
+
+ @Override
+ public Class<Transfer> getTypeClass() {
+ return Transfer.class;
+ }
+
+ @Override
+ public TypeEncoding<Transfer> getEncoding(Transfer transfer) {
+ return transferType.getEncoding(transfer);
+ }
+
+ @Override
+ public TypeEncoding<Transfer> getCanonicalEncoding() {
+ return transferType.getCanonicalEncoding();
+ }
+
+ @Override
+ public Collection<? extends TypeEncoding<Transfer>> getAllEncodings() {
+ return transferType.getAllEncodings();
+ }
+
+ @Override
+ public void write(Transfer value) {
+ WritableBuffer buffer = getEncoder().getBuffer();
+ int count = getElementCount(value);
+ byte encodingCode = deduceEncodingCode(value, count);
+
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ getEncoder().writeUnsignedLong(transferType.getDescriptor());
+
+ final int fieldWidth;
+
+ if (encodingCode == EncodingCodes.LIST8) {
+ fieldWidth = 1;
+ buffer.put(EncodingCodes.LIST8);
+ } else {
+ fieldWidth = 4;
+ buffer.put(EncodingCodes.LIST32);
+ }
+
+ int startIndex = buffer.position();
+
+ // Reserve space for the size and write the count of list elements.
+ if (fieldWidth == 1) {
+ buffer.put((byte) 0);
+ buffer.put((byte) count);
+ } else {
+ buffer.putInt(0);
+ buffer.putInt(count);
+ }
+
+ // Write the list elements and then compute total size written.
+ for (int i = 0; i < count; ++i) {
+ writeElement(value, i);
+ }
+
+ // Move back and write the size
+ int endIndex = buffer.position();
+ int writeSize = endIndex - startIndex - fieldWidth;
+
+ buffer.position(startIndex);
+ if (fieldWidth == 1) {
+ buffer.put((byte) writeSize);
+ } else {
+ buffer.putInt(writeSize);
+ }
+ buffer.position(endIndex);
+ }
+
+ private void writeElement(Transfer transfer, int index) {
+ switch (index) {
+ case 0:
+ getEncoder().writeUnsignedInteger(transfer.getHandle());
+ break;
+ case 1:
+ getEncoder().writeUnsignedInteger(transfer.getDeliveryId());
+ break;
+ case 2:
+ getEncoder().writeBinary(transfer.getDeliveryTag());
+ break;
+ case 3:
+ getEncoder().writeUnsignedInteger(transfer.getMessageFormat());
+ break;
+ case 4:
+ getEncoder().writeBoolean(transfer.getSettled());
+ break;
+ case 5:
+ getEncoder().writeBoolean(transfer.getMore());
+ break;
+ case 6:
+ ReceiverSettleMode rcvSettleMode = transfer.getRcvSettleMode();
+ getEncoder().writeObject(rcvSettleMode == null ? null : rcvSettleMode.getValue());
+ break;
+ case 7:
+ getEncoder().writeObject(transfer.getState());
+ break;
+ case 8:
+ getEncoder().writeBoolean(transfer.getResume());
+ break;
+ case 9:
+ getEncoder().writeBoolean(transfer.getAborted());
+ break;
+ case 10:
+ getEncoder().writeBoolean(transfer.getBatchable());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown Transfer value index: " + index);
+ }
+ }
+
+ private byte deduceEncodingCode(Transfer value, int elementCount) {
+ if (value.getState() != null) {
+ return EncodingCodes.LIST32;
+ } else if (value.getDeliveryTag() != null && value.getDeliveryTag().getLength() > 200) {
+ return EncodingCodes.LIST32;
+ } else {
+ return EncodingCodes.LIST8;
+ }
+ }
+
+ private int getElementCount(Transfer transfer) {
+ if (transfer.getBatchable()) {
+ return 11;
+ } else if (transfer.getAborted()) {
+ return 10;
+ } else if (transfer.getResume()) {
+ return 9;
+ } else if (transfer.getState() != null) {
+ return 8;
+ } else if (transfer.getRcvSettleMode() != null) {
+ return 7;
+ } else if (transfer.getMore()) {
+ return 6;
+ } else if (transfer.getSettled() != null) {
+ return 5;
+ } else if (transfer.getMessageFormat() != null) {
+ return 4;
+ } else if (transfer.getDeliveryTag() != null) {
+ return 3;
+ } else if (transfer.getDeliveryId() != null) {
+ return 2;
+ } else {
+ return 1;
+ }
+ }
+
+ public static void register(Decoder decoder, EncoderImpl encoder) {
+ FastPathTransferType type = new FastPathTransferType(encoder);
+ for(Object descriptor : DESCRIPTORS)
+ {
+ decoder.register(descriptor, (FastPathDescribedTypeConstructor<?>) type);
+ }
+ encoder.register(type);
+ }
+}
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FlowType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FlowType.java
index f9e91dc..499a397 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FlowType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/FlowType.java
@@ -36,7 +36,6 @@
import org.apache.qpid.proton.codec.DescribedTypeConstructor;
import org.apache.qpid.proton.codec.EncoderImpl;
-
public final class FlowType extends AbstractDescribedType<Flow,List> implements DescribedTypeConstructor<Flow>
{
private static final Object[] DESCRIPTORS =
@@ -46,7 +45,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000013L);
- private FlowType(EncoderImpl encoder)
+ FlowType(EncoderImpl encoder)
{
super(encoder);
}
@@ -176,7 +175,6 @@
return Flow.class;
}
-
public static void register(Decoder decoder, EncoderImpl encoder)
{
FlowType type = new FlowType(encoder);
@@ -186,6 +184,4 @@
}
encoder.register(type);
}
-
}
-
\ No newline at end of file
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/TransferType.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/TransferType.java
index 4ddbd49..6d18a07 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/TransferType.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/transport/TransferType.java
@@ -39,7 +39,6 @@
import org.apache.qpid.proton.codec.DescribedTypeConstructor;
import org.apache.qpid.proton.codec.EncoderImpl;
-
public final class TransferType extends AbstractDescribedType<Transfer,List> implements DescribedTypeConstructor<Transfer>
{
private static final Object[] DESCRIPTORS =
@@ -49,7 +48,7 @@
private static final UnsignedLong DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000014L);
- private TransferType(EncoderImpl encoder)
+ TransferType(EncoderImpl encoder)
{
super(encoder);
}
@@ -190,9 +189,6 @@
return Transfer.class;
}
-
-
-
public static void register(Decoder decoder, EncoderImpl encoder)
{
TransferType type = new TransferType(encoder);
@@ -202,6 +198,4 @@
}
encoder.register(type);
}
-
}
-
\ No newline at end of file
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/AmqpValueTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/AmqpValueTypeCodecTest.java
new file mode 100644
index 0000000..f2154e3
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/AmqpValueTypeCodecTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.UUID;
+
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
+import org.junit.Test;
+
+/**
+ * Test for decoder of the AmqpValue type.
+ */
+public class AmqpValueTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeAmqpValueString() throws IOException {
+ doTestDecodeAmqpValueSeries(1, new AmqpValue("test"));
+ }
+
+ @Test
+ public void testDecodeAmqpValueNull() throws IOException {
+ doTestDecodeAmqpValueSeries(1, new AmqpValue(null));
+ }
+
+ @Test
+ public void testDecodeAmqpValueUUID() throws IOException {
+ doTestDecodeAmqpValueSeries(1, new AmqpValue(UUID.randomUUID()));
+ }
+
+ @Test
+ public void testDecodeSmallSeriesOfAmqpValue() throws IOException {
+ doTestDecodeAmqpValueSeries(SMALL_SIZE, new AmqpValue("test"));
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfAmqpValue() throws IOException {
+ doTestDecodeAmqpValueSeries(LARGE_SIZE, new AmqpValue("test"));
+ }
+
+ private void doTestDecodeAmqpValueSeries(int size, AmqpValue value) throws IOException {
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(value);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof AmqpValue);
+
+ AmqpValue decoded = (AmqpValue) result;
+
+ assertEquals(value.getValue(), decoded.getValue());
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/ApplicationPropertiesCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/ApplicationPropertiesCodecTest.java
new file mode 100644
index 0000000..66f15bd
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/ApplicationPropertiesCodecTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
+import org.junit.Test;
+
+public class ApplicationPropertiesCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeSmallSeriesOfApplicationProperties() throws IOException {
+ doTestDecodeHeaderSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfApplicationProperties() throws IOException {
+ doTestDecodeHeaderSeries(LARGE_SIZE);
+ }
+
+ private void doTestDecodeHeaderSeries(int size) throws IOException {
+
+ Map<String, Object> propertiesMap = new LinkedHashMap<>();
+ ApplicationProperties properties = new ApplicationProperties(propertiesMap);
+
+ propertiesMap.put("key-1", "1");
+ propertiesMap.put("key-2", "2");
+ propertiesMap.put("key-3", "3");
+ propertiesMap.put("key-4", "4");
+ propertiesMap.put("key-5", "5");
+ propertiesMap.put("key-6", "6");
+ propertiesMap.put("key-7", "7");
+ propertiesMap.put("key-8", "8");
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(properties);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof ApplicationProperties);
+
+ ApplicationProperties decoded = (ApplicationProperties) result;
+
+ assertEquals(8, decoded.getValue().size());
+ assertTrue(decoded.getValue().equals(propertiesMap));
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/ApplicationPropertiesTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/ApplicationPropertiesTypeTest.java
new file mode 100644
index 0000000..6f11956
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/ApplicationPropertiesTypeTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
+import org.junit.Test;
+
+public class ApplicationPropertiesTypeTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeSmallSeriesOfApplicationProperties() throws IOException {
+ doTestDecodeHeaderSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfApplicationProperties() throws IOException {
+ doTestDecodeHeaderSeries(LARGE_SIZE);
+ }
+
+ private void doTestDecodeHeaderSeries(int size) throws IOException {
+
+ Map<String, Object> propertiesMap = new LinkedHashMap<>();
+ ApplicationProperties properties = new ApplicationProperties(propertiesMap);
+
+ long currentTime = System.currentTimeMillis();
+
+ propertiesMap.put("long-1", currentTime);
+ propertiesMap.put("date-1", new Date(currentTime));
+ propertiesMap.put("long-2", currentTime + 100);
+ propertiesMap.put("date-2", new Date(currentTime + 100));
+ propertiesMap.put("key-1", "1");
+ propertiesMap.put("key-2", "2");
+ propertiesMap.put("key-3", "3");
+ propertiesMap.put("key-4", "4");
+ propertiesMap.put("key-5", "5");
+ propertiesMap.put("key-6", "6");
+ propertiesMap.put("key-7", "7");
+ propertiesMap.put("key-8", "8");
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(properties);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof ApplicationProperties);
+
+ ApplicationProperties decoded = (ApplicationProperties) result;
+
+ assertEquals(properties.getValue().size(), decoded.getValue().size());
+ assertTrue(decoded.getValue().equals(propertiesMap));
+
+ assertEquals(currentTime, decoded.getValue().get("long-1"));
+ assertEquals(new Date(currentTime), decoded.getValue().get("date-1"));
+ assertEquals(currentTime + 100, decoded.getValue().get("long-2"));
+ assertEquals(new Date(currentTime + 100), decoded.getValue().get("date-2"));
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/ArrayTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/ArrayTypeCodecTest.java
new file mode 100644
index 0000000..2411b7e
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/ArrayTypeCodecTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.junit.Test;
+
+/**
+ * Test decoding of AMQP Array types
+ */
+public class ArrayTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_ARRAY_SIZE = 2048;
+ private final int SMALL_ARRAY_SIZE = 32;
+
+ @Test
+ public void testArrayOfPrimitiveBooleanObjects() throws IOException {
+ final int size = 10;
+
+ boolean[] source = new boolean[size];
+ for (int i = 0; i < size; ++i) {
+ source[i] = i % 2 == 0;
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject(buffer);
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+ assertTrue(result.getClass().getComponentType().isPrimitive());
+
+ boolean[] array = (boolean[]) result;
+ assertEquals(size, array.length);
+
+ for (int i = 0; i < size; ++i) {
+ assertEquals(source[i], array[i]);
+ }
+ }
+
+ @Test
+ public void testZeroSizedArrayOfPrimitiveBooleanObjects() throws IOException {
+ boolean[] source = new boolean[0];
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+ assertTrue(result.getClass().getComponentType().isPrimitive());
+
+ boolean[] array = (boolean[]) result;
+ assertEquals(source.length, array.length);
+ }
+
+ @Test
+ public void testArrayOfBooleanObjects() throws IOException {
+ final int size = 10;
+
+ Boolean[] source = new Boolean[size];
+ for (int i = 0; i < size; ++i) {
+ source[i] = i % 2 == 0;
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+ assertTrue(result.getClass().getComponentType().isPrimitive());
+
+ boolean[] array = (boolean[]) result;
+ assertEquals(size, array.length);
+
+ for (int i = 0; i < size; ++i) {
+ assertEquals(source[i], array[i]);
+ }
+ }
+
+ @Test
+ public void testZeroSizedArrayOfBooleanObjects() throws IOException {
+ Boolean[] source = new Boolean[0];
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+ assertTrue(result.getClass().getComponentType().isPrimitive());
+
+ boolean[] array = (boolean[]) result;
+ assertEquals(source.length, array.length);
+ }
+
+ @Test
+ public void testDecodeSmallBooleanArray() throws IOException {
+ doTestDecodeBooleanArrayType(SMALL_ARRAY_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeBooleanArray() throws IOException {
+ doTestDecodeBooleanArrayType(LARGE_ARRAY_SIZE);
+ }
+
+ private void doTestDecodeBooleanArrayType(int size) throws IOException {
+ boolean[] source = new boolean[size];
+ for (int i = 0; i < size; ++i) {
+ source[i] = i % 2 == 0;
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+ assertTrue(result.getClass().getComponentType().isPrimitive());
+
+ boolean[] array = (boolean[]) result;
+ assertEquals(size, array.length);
+
+ for (int i = 0; i < size; ++i) {
+ assertEquals(source[i], array[i]);
+ }
+ }
+
+ @Test
+ public void testDecodeSmallSymbolArray() throws IOException {
+ doTestDecodeSymbolArrayType(SMALL_ARRAY_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSymbolArray() throws IOException {
+ doTestDecodeSymbolArrayType(LARGE_ARRAY_SIZE);
+ }
+
+ private void doTestDecodeSymbolArrayType(int size) throws IOException {
+ Symbol[] source = new Symbol[size];
+ for (int i = 0; i < size; ++i) {
+ source[i] = Symbol.valueOf("test->" + i);
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+
+ Symbol[] array = (Symbol[]) result;
+ assertEquals(size, array.length);
+
+ for (int i = 0; i < size; ++i) {
+ assertEquals(source[i], array[i]);
+ }
+ }
+
+ @Test
+ public void testDecodeSmallUUIDArray() throws IOException {
+ doTestDecodeUUDIArrayType(SMALL_ARRAY_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeUUDIArray() throws IOException {
+ doTestDecodeUUDIArrayType(LARGE_ARRAY_SIZE);
+ }
+
+ private void doTestDecodeUUDIArrayType(int size) throws IOException {
+ UUID[] source = new UUID[size];
+ for (int i = 0; i < size; ++i) {
+ source[i] = UUID.randomUUID();
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+
+ UUID[] array = (UUID[]) result;
+ assertEquals(size, array.length);
+
+ for (int i = 0; i < size; ++i) {
+ assertEquals(source[i], array[i]);
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testArrayOfListsOfUUIDs() throws IOException {
+ ArrayList<UUID>[] source = new ArrayList[2];
+ for (int i = 0; i < source.length; ++i) {
+ source[i] = new ArrayList<UUID>(3);
+ source[i].add(UUID.randomUUID());
+ source[i].add(UUID.randomUUID());
+ source[i].add(UUID.randomUUID());
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+
+ List[] list = (List[]) result;
+ assertEquals(source.length, list.length);
+
+ for (int i = 0; i < list.length; ++i) {
+ assertEquals(source[i], list[i]);
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testArrayOfMApsOfStringToUUIDs() throws IOException {
+ Map<String, UUID>[] source = new LinkedHashMap[2];
+ for (int i = 0; i < source.length; ++i) {
+ source[i] = new LinkedHashMap<String, UUID>();
+ source[i].put("1", UUID.randomUUID());
+ source[i].put("2", UUID.randomUUID());
+ source[i].put("3", UUID.randomUUID());
+ }
+
+ encoder.writeArray(source);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+
+ Map[] map = (Map[]) result;
+ assertEquals(source.length, map.length);
+
+ for (int i = 0; i < map.length; ++i) {
+ assertEquals(source[i], map[i]);
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/Benchmark.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/Benchmark.java
new file mode 100644
index 0000000..529a808
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/Benchmark.java
@@ -0,0 +1,420 @@
+/*
+ *
+ * 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.proton.codec;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedByte;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
+import org.apache.qpid.proton.amqp.UnsignedShort;
+import org.apache.qpid.proton.amqp.messaging.Accepted;
+import org.apache.qpid.proton.amqp.messaging.Data;
+import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
+import org.apache.qpid.proton.amqp.messaging.Header;
+import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
+import org.apache.qpid.proton.amqp.messaging.Properties;
+import org.apache.qpid.proton.amqp.transport.Disposition;
+import org.apache.qpid.proton.amqp.transport.Flow;
+import org.apache.qpid.proton.amqp.transport.Role;
+import org.apache.qpid.proton.amqp.transport.Transfer;
+
+public class Benchmark implements Runnable {
+
+ private static final int ITERATIONS = 10 * 1024 * 1024;
+
+ private ByteBuffer byteBuf = ByteBuffer.allocate(8192);
+ private BenchmarkResult resultSet = new BenchmarkResult();
+ private boolean warming = true;
+
+ private final DecoderImpl decoder = new DecoderImpl();
+ private final EncoderImpl encoder = new EncoderImpl(decoder);
+
+ public static final void main(String[] args) throws IOException, InterruptedException {
+ System.out.println("Current PID: " + ManagementFactory.getRuntimeMXBean().getName());
+ Benchmark benchmark = new Benchmark();
+ benchmark.run();
+ }
+
+ @Override
+ public void run() {
+ AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+
+ encoder.setByteBuffer(byteBuf);
+ decoder.setByteBuffer(byteBuf);
+
+ try {
+ doBenchmarks();
+ warming = false;
+ doBenchmarks();
+ } catch (IOException e) {
+ System.out.println("Unexpected error: " + e.getMessage());
+ }
+ }
+
+ private void time(String message, BenchmarkResult resultSet) {
+ if (!warming) {
+ System.out.println("Benchamrk of type: " + message + ": ");
+ System.out.println(" Encode time = " + resultSet.getEncodeTimeMills());
+ System.out.println(" Decode time = " + resultSet.getDecodeTimeMills());
+ }
+ }
+
+ private final void doBenchmarks() throws IOException {
+ benchmarkListOfInts();
+ benchmarkUUIDs();
+ benchmarkHeader();
+ benchmarkProperties();
+ benchmarkMessageAnnotations();
+ benchmarkApplicationProperties();
+ benchmarkSymbols();
+ benchmarkTransfer();
+ benchmarkFlow();
+ benchmarkDisposition();
+ benchmarkString();
+ benchmarkData();
+ warming = false;
+ }
+
+ private void benchmarkListOfInts() throws IOException {
+ ArrayList<Object> list = new ArrayList<>(10);
+ for (int j = 0; j < 10; j++) {
+ list.add(0);
+ }
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeList(list);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readList();
+ }
+ resultSet.decodesComplete();
+
+ time("List<Integer>", resultSet);
+ }
+
+ private void benchmarkUUIDs() throws IOException {
+ UUID uuid = UUID.randomUUID();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeUUID(uuid);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readUUID();
+ }
+ resultSet.decodesComplete();
+
+ time("UUID", resultSet);
+ }
+
+ private void benchmarkHeader() throws IOException {
+ Header header = new Header();
+ header.setDurable(true);
+ header.setFirstAcquirer(true);
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(header);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("Header", resultSet);
+ }
+
+ private void benchmarkTransfer() throws IOException {
+ Transfer transfer = new Transfer();
+ transfer.setDeliveryTag(new Binary(new byte[] {1, 2, 3}));
+ transfer.setHandle(UnsignedInteger.valueOf(10));
+ transfer.setMessageFormat(UnsignedInteger.ZERO);
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(transfer);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("Transfer", resultSet);
+ }
+
+ private void benchmarkFlow() throws IOException {
+ Flow flow = new Flow();
+ flow.setNextIncomingId(UnsignedInteger.valueOf(1));
+ flow.setIncomingWindow(UnsignedInteger.valueOf(2047));
+ flow.setNextOutgoingId(UnsignedInteger.valueOf(1));
+ flow.setOutgoingWindow(UnsignedInteger.MAX_VALUE);
+ flow.setHandle(UnsignedInteger.ZERO);
+ flow.setDeliveryCount(UnsignedInteger.valueOf(10));
+ flow.setLinkCredit(UnsignedInteger.valueOf(1000));
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(flow);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("Flow", resultSet);
+ }
+
+ private void benchmarkProperties() throws IOException {
+ Properties properties = new Properties();
+ properties.setTo("queue:1");
+ properties.setMessageId("ID:Message:1");
+ properties.setCreationTime(new Date(System.currentTimeMillis()));
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(properties);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("Properties", resultSet);
+ }
+
+ private void benchmarkMessageAnnotations() throws IOException {
+ MessageAnnotations annotations = new MessageAnnotations(new HashMap<Symbol, Object>());
+ annotations.getValue().put(Symbol.valueOf("test1"), UnsignedByte.valueOf((byte) 128));
+ annotations.getValue().put(Symbol.valueOf("test2"), UnsignedShort.valueOf((short) 128));
+ annotations.getValue().put(Symbol.valueOf("test3"), UnsignedInteger.valueOf((byte) 128));
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(annotations);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("MessageAnnotations", resultSet);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void benchmarkApplicationProperties() throws IOException {
+ ApplicationProperties properties = new ApplicationProperties(new HashMap<String, Object>());
+ properties.getValue().put("test1", UnsignedByte.valueOf((byte) 128));
+ properties.getValue().put("test2", UnsignedShort.valueOf((short) 128));
+ properties.getValue().put("test3", UnsignedInteger.valueOf((byte) 128));
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(properties);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("ApplicationProperties", resultSet);
+ }
+
+ private void benchmarkSymbols() throws IOException {
+ Symbol symbol1 = Symbol.valueOf("Symbol-1");
+ Symbol symbol2 = Symbol.valueOf("Symbol-2");
+ Symbol symbol3 = Symbol.valueOf("Symbol-3");
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeSymbol(symbol1);
+ encoder.writeSymbol(symbol2);
+ encoder.writeSymbol(symbol3);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readSymbol();
+ decoder.readSymbol();
+ decoder.readSymbol();
+ }
+ resultSet.decodesComplete();
+
+ time("Symbol", resultSet);
+ }
+
+ private void benchmarkDisposition() throws IOException {
+ Disposition disposition = new Disposition();
+ disposition.setRole(Role.RECEIVER);
+ disposition.setSettled(true);
+ disposition.setState(Accepted.getInstance());
+ disposition.setFirst(UnsignedInteger.valueOf(2));
+ disposition.setLast(UnsignedInteger.valueOf(2));
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(disposition);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("Disposition", resultSet);
+ }
+
+ private void benchmarkString() throws IOException {
+ String string1 = new String("String-1");
+ String string2 = new String("String-2");
+ String string3 = new String("String-3");
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeString(string1);
+ encoder.writeString(string2);
+ encoder.writeString(string3);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readString();
+ decoder.readString();
+ decoder.readString();
+ }
+ resultSet.decodesComplete();
+
+ time("String", resultSet);
+ }
+
+ private void benchmarkData() throws IOException {
+ Data data1 = new Data(new Binary(new byte[] {1, 2, 3}));
+ Data data2 = new Data(new Binary(new byte[] {4, 5, 6}));
+ Data data3 = new Data(new Binary(new byte[] {7, 8, 9}));
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.clear();
+ encoder.writeObject(data1);
+ encoder.writeObject(data2);
+ encoder.writeObject(data3);
+ }
+ resultSet.encodesComplete();
+
+ resultSet.start();
+ for (int i = 0; i < ITERATIONS; i++) {
+ byteBuf.flip();
+ decoder.readObject();
+ decoder.readObject();
+ decoder.readObject();
+ }
+ resultSet.decodesComplete();
+
+ time("Data", resultSet);
+ }
+
+ private static class BenchmarkResult {
+
+ private long startTime;
+
+ private long encodeTime;
+ private long decodeTime;
+
+ public void start() {
+ startTime = System.nanoTime();
+ }
+
+ public void encodesComplete() {
+ encodeTime = System.nanoTime() - startTime;
+ }
+
+ public void decodesComplete() {
+ decodeTime = System.nanoTime() - startTime;
+ }
+
+ public long getEncodeTimeMills() {
+ return TimeUnit.NANOSECONDS.toMillis(encodeTime);
+ }
+
+ public long getDecodeTimeMills() {
+ return TimeUnit.NANOSECONDS.toMillis(decodeTime);
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/CodecTestSupport.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/CodecTestSupport.java
new file mode 100644
index 0000000..bb33697
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/CodecTestSupport.java
@@ -0,0 +1,48 @@
+/*
+ * 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.proton.codec;
+
+import java.nio.ByteBuffer;
+
+import org.junit.Before;
+
+/**
+ * Support class for tests of the type decoders
+ */
+public class CodecTestSupport {
+
+ public static final int DEFAULT_MAX_BUFFER = 256 * 1024;
+
+ final DecoderImpl decoder = new DecoderImpl();
+ final EncoderImpl encoder = new EncoderImpl(decoder);
+
+ ByteBuffer buffer;
+
+ @Before
+ public void setUp() {
+ AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+
+ buffer = ByteBuffer.allocate(getMaxBufferSize());
+
+ encoder.setByteBuffer(buffer);
+ decoder.setByteBuffer(buffer);
+ }
+
+ public int getMaxBufferSize() {
+ return DEFAULT_MAX_BUFFER;
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/HeaderTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/HeaderTypeCodecTest.java
new file mode 100644
index 0000000..3b5d93c
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/HeaderTypeCodecTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+
+import org.apache.qpid.proton.amqp.UnsignedByte;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
+import org.apache.qpid.proton.amqp.messaging.Header;
+import org.junit.Test;
+
+/**
+ * Test for decoder of AMQP Header type.
+ */
+public class HeaderTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024 * 10;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeHeader() throws IOException {
+ doTestDecodeHeaderSeries(1);
+ }
+
+ @Test
+ public void testDecodeSmallSeriesOfHeaders() throws IOException {
+ doTestDecodeHeaderSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfHeaders() throws IOException {
+ doTestDecodeHeaderSeries(LARGE_SIZE);
+ }
+
+ private void doTestDecodeHeaderSeries(int size) throws IOException {
+ Header header = new Header();
+
+ header.setDurable(Boolean.TRUE);
+ header.setPriority(UnsignedByte.valueOf((byte) 3));
+ header.setDeliveryCount(UnsignedInteger.valueOf(10));
+ header.setFirstAcquirer(Boolean.FALSE);
+ header.setTtl(UnsignedInteger.valueOf(500));
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(header);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof Header);
+
+ Header decoded = (Header) result;
+
+ assertEquals(3, decoded.getPriority().intValue());
+ assertTrue(decoded.getDurable().booleanValue());
+ }
+ }
+
+ @Test
+ public void testSkipHeader() throws IOException {
+ Header header1 = new Header();
+ Header header2 = new Header();
+
+ header1.setDurable(Boolean.FALSE);
+ header2.setDurable(Boolean.TRUE);
+
+ encoder.writeObject(header1);
+ encoder.writeObject(header2);
+
+ buffer.clear();
+
+ TypeConstructor<?> headerType = decoder.readConstructor();
+ assertEquals(Header.class, headerType.getTypeClass());
+ headerType.skipValue();
+
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof Header);
+
+ Header decoded = (Header) result;
+ assertTrue(decoded.getDurable().booleanValue());
+ }
+
+ @Test
+ public void testDecodeHeaderArray() throws IOException {
+ Header header = new Header();
+
+ header.setDurable(Boolean.TRUE);
+ header.setPriority(UnsignedByte.valueOf((byte) 3));
+ header.setDeliveryCount(UnsignedInteger.valueOf(10));
+ header.setFirstAcquirer(Boolean.FALSE);
+ header.setTtl(UnsignedInteger.valueOf(500));
+
+ Header[] source = new Header[32];
+
+ for (int i = 0; i < source.length; ++i) {
+ source[i] = header;
+ }
+
+ encoder.writeObject(source);
+
+ buffer.clear();
+
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result.getClass().isArray());
+
+ final Object[] resultArray = (Object[]) result;
+
+ for (int i = 0; i < source.length; ++i) {
+
+ assertTrue(resultArray[i] instanceof Header);
+
+ Header decoded = (Header) resultArray[i];
+
+ assertEquals(3, decoded.getPriority().intValue());
+ assertTrue(decoded.getDurable().booleanValue());
+ assertEquals(header.getDeliveryCount(), decoded.getDeliveryCount());
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/ListTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/ListTypeCodecTest.java
new file mode 100644
index 0000000..6b5ad5d
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/ListTypeCodecTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
+import org.junit.Test;
+
+/**
+ * Test for the Proton List encoder / decoder
+ */
+public class ListTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeSmallSeriesOfLists() throws IOException {
+ doTestDecodeListSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfLists() throws IOException {
+ doTestDecodeListSeries(LARGE_SIZE);
+ }
+
+ @Test
+ public void testDecodeSmallSeriesOfSymbolLists() throws IOException {
+ doTestDecodeSymbolListSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfSymbolLists() throws IOException {
+ doTestDecodeSymbolListSeries(LARGE_SIZE);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doTestDecodeSymbolListSeries(int size) throws IOException {
+ List<Object> list = new ArrayList<>();
+
+ for (int i = 0; i < 50; ++i) {
+ list.add(Symbol.valueOf(String.valueOf(i)));
+ }
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(list);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof List);
+
+ List<Object> resultList = (List<Object>) result;
+
+ assertEquals(list.size(), resultList.size());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doTestDecodeListSeries(int size) throws IOException {
+ List<Object> list = new ArrayList<>();
+
+ Date timeNow = new Date(System.currentTimeMillis());
+
+ list.add("ID:Message-1:1:1:0");
+ list.add(new Binary(new byte[1]));
+ list.add("queue:work");
+ list.add(Symbol.valueOf("text/UTF-8"));
+ list.add(Symbol.valueOf("text"));
+ list.add(timeNow);
+ list.add(UnsignedInteger.valueOf(1));
+ list.add(UUID.randomUUID());
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(list);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof List);
+
+ List<Object> resultList = (List<Object>) result;
+
+ assertEquals(list.size(), resultList.size());
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/MapTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/MapTypeCodecTest.java
new file mode 100644
index 0000000..32a0119
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/MapTypeCodecTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.junit.Test;
+
+public class MapTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeSmallSeriesOfMaps() throws IOException {
+ doTestDecodeMapSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfMaps() throws IOException {
+ doTestDecodeMapSeries(LARGE_SIZE);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doTestDecodeMapSeries(int size) throws IOException {
+
+ String myBoolKey = "myBool";
+ boolean myBool = true;
+ String myByteKey = "myByte";
+ byte myByte = 4;
+ String myBytesKey = "myBytes";
+ byte[] myBytes = myBytesKey.getBytes();
+ String myCharKey = "myChar";
+ char myChar = 'd';
+ String myDoubleKey = "myDouble";
+ double myDouble = 1234567890123456789.1234;
+ String myFloatKey = "myFloat";
+ float myFloat = 1.1F;
+ String myIntKey = "myInt";
+ int myInt = Integer.MAX_VALUE;
+ String myLongKey = "myLong";
+ long myLong = Long.MAX_VALUE;
+ String myShortKey = "myShort";
+ short myShort = 25;
+ String myStringKey = "myString";
+ String myString = myStringKey;
+
+ Map<String, Object> map = new LinkedHashMap<String, Object>();
+ map.put(myBoolKey, myBool);
+ map.put(myByteKey, myByte);
+ map.put(myBytesKey, new Binary(myBytes));
+ map.put(myCharKey, myChar);
+ map.put(myDoubleKey, myDouble);
+ map.put(myFloatKey, myFloat);
+ map.put(myIntKey, myInt);
+ map.put(myLongKey, myLong);
+ map.put(myShortKey, myShort);
+ map.put(myStringKey, myString);
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(map);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof Map);
+
+ Map<String, Object> resultMap = (Map<String, Object>) result;
+
+ assertEquals(map.size(), resultMap.size());
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/MessageAnnotationsTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/MessageAnnotationsTypeCodecTest.java
new file mode 100644
index 0000000..d1f6ba7
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/MessageAnnotationsTypeCodecTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedByte;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
+import org.apache.qpid.proton.amqp.UnsignedShort;
+import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
+import org.junit.Test;
+
+public class MessageAnnotationsTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeSmallSeriesOfMessageAnnotations() throws IOException {
+ doTestDecodeMessageAnnotationsSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfMessageAnnotations() throws IOException {
+ doTestDecodeMessageAnnotationsSeries(LARGE_SIZE);
+ }
+
+ @Test
+ public void testDecodeLMessageAnnotations() throws IOException {
+ doTestDecodeMessageAnnotationsSeries(1);
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private void doTestDecodeMessageAnnotationsSeries(int size) throws IOException {
+
+ final Symbol SYMBOL_1 = Symbol.valueOf("test1");
+ final Symbol SYMBOL_2 = Symbol.valueOf("test2");
+ final Symbol SYMBOL_3 = Symbol.valueOf("test3");
+
+ MessageAnnotations annotations = new MessageAnnotations(new HashMap());
+ annotations.getValue().put(SYMBOL_1, UnsignedByte.valueOf((byte) 128));
+ annotations.getValue().put(SYMBOL_2, UnsignedShort.valueOf((short) 128));
+ annotations.getValue().put(SYMBOL_3, UnsignedInteger.valueOf(128));
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(annotations);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof MessageAnnotations);
+
+ MessageAnnotations readAnnotations = (MessageAnnotations) result;
+
+ Map<Symbol, Object> resultMap = readAnnotations.getValue();
+
+ assertEquals(annotations.getValue().size(), resultMap.size());
+ assertEquals(resultMap.get(SYMBOL_1), UnsignedByte.valueOf((byte) 128));
+ assertEquals(resultMap.get(SYMBOL_2), UnsignedShort.valueOf((short) 128));
+ assertEquals(resultMap.get(SYMBOL_3), UnsignedInteger.valueOf(128));
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/NoLocalType.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/NoLocalType.java
new file mode 100644
index 0000000..4ec6724
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/NoLocalType.java
@@ -0,0 +1,46 @@
+/*
+ * 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.proton.codec;
+
+import org.apache.qpid.proton.amqp.DescribedType;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+
+/**
+ * A Described Type wrapper for JMS no local option for MessageConsumer.
+ */
+public class NoLocalType implements DescribedType {
+
+ public static final NoLocalType NO_LOCAL = new NoLocalType();
+
+ public static final UnsignedLong DESCRIPTOR_CODE = UnsignedLong.valueOf(0x0000468C00000003L);
+
+ private final String noLocal;
+
+ public NoLocalType() {
+ this.noLocal = "NoLocalFilter{}";
+ }
+
+ @Override
+ public Object getDescriptor() {
+ return DESCRIPTOR_CODE;
+ }
+
+ @Override
+ public String getDescribed() {
+ return this.noLocal;
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/PropertiesCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/PropertiesCodecTest.java
new file mode 100644
index 0000000..e6ef140
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/PropertiesCodecTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.Date;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
+import org.apache.qpid.proton.amqp.messaging.Properties;
+import org.junit.Test;
+
+/**
+ * Test for decoder of AMQP Properties type.
+ */
+public class PropertiesCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeSmallSeriesOfPropertiess() throws IOException {
+ doTestDecodePropertiesSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfPropertiess() throws IOException {
+ doTestDecodePropertiesSeries(LARGE_SIZE);
+ }
+
+ private void doTestDecodePropertiesSeries(int size) throws IOException {
+ Properties properties = new Properties();
+
+ Date timeNow = new Date(System.currentTimeMillis());
+
+ properties.setMessageId("ID:Message-1:1:1:0");
+ properties.setUserId(new Binary(new byte[1]));
+ properties.setTo("queue:work");
+ properties.setSubject("help");
+ properties.setReplyTo("queue:temp:me");
+ properties.setContentEncoding(Symbol.valueOf("text/UTF-8"));
+ properties.setContentType(Symbol.valueOf("text"));
+ properties.setCorrelationId("correlation-id");
+ properties.setAbsoluteExpiryTime(timeNow);
+ properties.setCreationTime(timeNow);
+ properties.setGroupId("group-1");
+ properties.setGroupSequence(UnsignedInteger.valueOf(1));
+ properties.setReplyToGroupId("group-1");
+
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(properties);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof Properties);
+
+ Properties decoded = (Properties) result;
+
+ assertNotNull(decoded.getAbsoluteExpiryTime());
+ assertEquals(timeNow, decoded.getAbsoluteExpiryTime());
+ assertEquals(Symbol.valueOf("text/UTF-8"), decoded.getContentEncoding());
+ assertEquals(Symbol.valueOf("text"), decoded.getContentType());
+ assertEquals("correlation-id", decoded.getCorrelationId());
+ assertEquals(timeNow, decoded.getCreationTime());
+ assertEquals("group-1", decoded.getGroupId());
+ assertEquals(UnsignedInteger.valueOf(1), decoded.getGroupSequence());
+ assertEquals("ID:Message-1:1:1:0", decoded.getMessageId());
+ assertEquals("queue:temp:me", decoded.getReplyTo());
+ assertEquals("group-1", decoded.getReplyToGroupId());
+ assertEquals("help", decoded.getSubject());
+ assertEquals("queue:work", decoded.getTo());
+ assertTrue(decoded.getUserId() instanceof Binary);
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/StringTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/StringTypeTest.java
index 90cfe26..6c376d5 100644
--- a/proton-j/src/test/java/org/apache/qpid/proton/codec/StringTypeTest.java
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/StringTypeTest.java
@@ -125,6 +125,30 @@
}
}
+ @Test
+ public void testSkipString()
+ {
+ final DecoderImpl decoder = new DecoderImpl();
+ final EncoderImpl encoder = new EncoderImpl(decoder);
+ AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+ final ByteBuffer buffer = ByteBuffer.allocate(64);
+
+ decoder.setByteBuffer(buffer);
+ encoder.setByteBuffer(buffer);
+
+ encoder.writeString("skipped");
+ encoder.writeString("read");
+
+ buffer.clear();
+
+ TypeConstructor<?> stringType = decoder.readConstructor();
+ assertEquals(String.class, stringType.getTypeClass());
+ stringType.skipValue();
+
+ String result = decoder.readString();
+ assertEquals("read", result);
+ }
+
// build up some test data with a set of suitable Unicode characters
private static List<String> generateTestData()
{
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java
new file mode 100644
index 0000000..8e2d088
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.*;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
+import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
+import org.apache.qpid.proton.amqp.transport.Transfer;
+import org.junit.Test;
+
+/**
+ * Tests for encode / decode of Transfer types
+ */
+public class TransferTypeTest extends CodecTestSupport {
+
+ @Test
+ public void testEncodeDecodeTransfers() {
+ Transfer transfer = new Transfer();
+ transfer.setHandle(UnsignedInteger.ONE);
+ transfer.setDeliveryTag(new Binary(new byte[] {0, 1}));
+ transfer.setMessageFormat(UnsignedInteger.ZERO);
+ transfer.setDeliveryId(UnsignedInteger.valueOf(127));
+ transfer.setAborted(false);
+ transfer.setBatchable(true);
+ transfer.setRcvSettleMode(ReceiverSettleMode.SECOND);
+
+ encoder.writeObject(transfer);
+ buffer.clear();
+ final Transfer outputValue = (Transfer) decoder.readObject();
+
+ assertEquals(transfer.getHandle(), outputValue.getHandle());
+ assertEquals(transfer.getMessageFormat(), outputValue.getMessageFormat());
+ assertEquals(transfer.getDeliveryTag(), outputValue.getDeliveryTag());
+ assertEquals(transfer.getDeliveryId(), outputValue.getDeliveryId());
+ assertEquals(transfer.getAborted(), outputValue.getAborted());
+ assertEquals(transfer.getBatchable(), outputValue.getBatchable());
+ assertEquals(transfer.getRcvSettleMode(), outputValue.getRcvSettleMode());
+ }
+
+ @Test
+ public void testSkipValue() {
+ Transfer transfer = new Transfer();
+ transfer.setHandle(UnsignedInteger.ONE);
+ transfer.setDeliveryTag(new Binary(new byte[] {0, 1}));
+ transfer.setMessageFormat(UnsignedInteger.ZERO);
+ transfer.setDeliveryId(UnsignedInteger.valueOf(127));
+ transfer.setAborted(false);
+ transfer.setBatchable(true);
+ transfer.setRcvSettleMode(ReceiverSettleMode.SECOND);
+
+ encoder.writeObject(transfer);
+
+ transfer.setHandle(UnsignedInteger.valueOf(2));
+
+ encoder.writeObject(transfer);
+
+ buffer.clear();
+
+ TypeConstructor<?> type = decoder.readConstructor();
+ assertEquals(Transfer.class, type.getTypeClass());
+ type.skipValue();
+
+ Transfer result = (Transfer) decoder.readObject();
+ assertEquals(UnsignedInteger.valueOf(2), result.getHandle());
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/UnknownDescribedTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/UnknownDescribedTypeCodecTest.java
new file mode 100644
index 0000000..2aaaac6
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/UnknownDescribedTypeCodecTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.proton.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.qpid.proton.amqp.DescribedType;
+import org.junit.Test;
+
+/**
+ * Tests the handling of UnknownDescribedType instances.
+ */
+public class UnknownDescribedTypeCodecTest extends CodecTestSupport {
+
+ private final int LARGE_SIZE = 1024;
+ private final int SMALL_SIZE = 32;
+
+ @Test
+ public void testDecodeUnknownDescribedType() throws Exception {
+ encoder.writeObject(NoLocalType.NO_LOCAL);
+
+ buffer.clear();
+
+ Object result = decoder.readObject();
+ assertTrue(result instanceof DescribedType);
+ DescribedType resultTye = (DescribedType) result;
+ assertEquals(NoLocalType.NO_LOCAL.getDescriptor(), resultTye.getDescriptor());
+ }
+
+ @Test
+ public void testDecodeSmallSeriesOfUnknownDescribedTypes() throws IOException {
+ doTestDecodeUnknownDescribedTypeSeries(SMALL_SIZE);
+ }
+
+ @Test
+ public void testDecodeLargeSeriesOfUnknownDescribedTypes() throws IOException {
+ doTestDecodeUnknownDescribedTypeSeries(LARGE_SIZE);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testUnknownDescribedTypeInList() throws IOException {
+ List<Object> listOfUnkowns = new ArrayList<>();
+
+ listOfUnkowns.add(NoLocalType.NO_LOCAL);
+
+ encoder.writeList(listOfUnkowns);
+
+ buffer.clear();
+
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof List);
+
+ final List<Object> decodedList = (List<Object>) result;
+ assertEquals(1, decodedList.size());
+
+ final Object listEntry = decodedList.get(0);
+ assertTrue(listEntry instanceof DescribedType);
+
+ DescribedType resultTye = (DescribedType) listEntry;
+ assertEquals(NoLocalType.NO_LOCAL.getDescriptor(), resultTye.getDescriptor());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testUnknownDescribedTypeInMap() throws IOException {
+ Map<Object, Object> mapOfUnknowns = new HashMap<>();
+
+ mapOfUnknowns.put(NoLocalType.NO_LOCAL.getDescriptor(), NoLocalType.NO_LOCAL);
+
+ encoder.writeMap(mapOfUnknowns);
+
+ buffer.clear();
+
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof Map);
+
+ final Map<Object, Object> decodedMap = (Map<Object, Object>) result;
+ assertEquals(1, decodedMap.size());
+
+ final Object mapEntry = decodedMap.get(NoLocalType.NO_LOCAL.getDescriptor());
+ assertTrue(mapEntry instanceof DescribedType);
+
+ DescribedType resultTye = (DescribedType) mapEntry;
+ assertEquals(NoLocalType.NO_LOCAL.getDescriptor(), resultTye.getDescriptor());
+ }
+
+ private void doTestDecodeUnknownDescribedTypeSeries(int size) throws IOException {
+ for (int i = 0; i < size; ++i) {
+ encoder.writeObject(NoLocalType.NO_LOCAL);
+ }
+
+ buffer.clear();
+
+ for (int i = 0; i < size; ++i) {
+ final Object result = decoder.readObject();
+
+ assertNotNull(result);
+ assertTrue(result instanceof DescribedType);
+
+ DescribedType resultTye = (DescribedType) result;
+ assertEquals(NoLocalType.NO_LOCAL.getDescriptor(), resultTye.getDescriptor());
+ }
+ }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedIntegerTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedIntegerTypeTest.java
index 45523e2..42b80cc 100644
--- a/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedIntegerTypeTest.java
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedIntegerTypeTest.java
@@ -22,6 +22,8 @@
import static org.junit.Assert.*;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.codec.UnsignedIntegerType.UnsignedIntegerEncoding;
import org.junit.Test;
@@ -63,4 +65,28 @@
UnsignedIntegerEncoding encoding = ult.getEncoding(UnsignedInteger.valueOf(val));
assertEquals("incorrect encoding returned", EncodingCodes.UINT, encoding.getEncodingCode());
}
+
+ @Test
+ public void testSkipValue()
+ {
+ final DecoderImpl decoder = new DecoderImpl();
+ final EncoderImpl encoder = new EncoderImpl(decoder);
+ AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+ final ByteBuffer buffer = ByteBuffer.allocate(64);
+
+ decoder.setByteBuffer(buffer);
+ encoder.setByteBuffer(buffer);
+
+ encoder.writeUnsignedInteger(UnsignedInteger.ZERO);
+ encoder.writeUnsignedInteger(UnsignedInteger.ONE);
+
+ buffer.clear();
+
+ TypeConstructor<?> type = decoder.readConstructor();
+ assertEquals(UnsignedInteger.class, type.getTypeClass());
+ type.skipValue();
+
+ UnsignedInteger result = decoder.readUnsignedInteger();
+ assertEquals(UnsignedInteger.ONE, result);
+ }
}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedLongTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedLongTypeTest.java
index bcc6ff8..7b4cea3 100644
--- a/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedLongTypeTest.java
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/UnsignedLongTypeTest.java
@@ -23,6 +23,7 @@
import static org.junit.Assert.*;
import java.math.BigInteger;
+import java.nio.ByteBuffer;
import org.apache.qpid.proton.amqp.UnsignedLong;
import org.apache.qpid.proton.codec.UnsignedLongType.UnsignedLongEncoding;
@@ -65,4 +66,28 @@
UnsignedLongEncoding encoding = ult.getEncoding(UnsignedLong.valueOf(bigInt));
assertEquals("incorrect encoding returned", EncodingCodes.ULONG, encoding.getEncodingCode());
}
+
+ @Test
+ public void testSkipValue()
+ {
+ final DecoderImpl decoder = new DecoderImpl();
+ final EncoderImpl encoder = new EncoderImpl(decoder);
+ AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+ final ByteBuffer buffer = ByteBuffer.allocate(64);
+
+ decoder.setByteBuffer(buffer);
+ encoder.setByteBuffer(buffer);
+
+ encoder.writeUnsignedLong(UnsignedLong.ZERO);
+ encoder.writeUnsignedLong(UnsignedLong.valueOf(1));
+
+ buffer.clear();
+
+ TypeConstructor<?> type = decoder.readConstructor();
+ assertEquals(UnsignedLong.class, type.getTypeClass());
+ type.skipValue();
+
+ UnsignedLong result = decoder.readUnsignedLong();
+ assertEquals(UnsignedLong.valueOf(1), result);
+ }
}
