PROTON-1941 Add ensureRemaining to the WritableBuffer interface

Provide an access point for the codec to request that the WritableBuffer
be sized to accept a given number of bytes usually in relation to larger
incoming write of a complex type.  Add default implementation  to preseve
backwards compatability and an implementation in the ByteBuffer wrapper
that throws early if nont enough bytes exist to handle the announced
pending writes.
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 f4f0c8a..06ece52 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
@@ -84,11 +84,13 @@
         decoder.register(this);
     }
 
+    @Override
     public Class<Object[]> getTypeClass()
     {
         return Object[].class;
     }
 
+    @Override
     public ArrayEncoding getEncoding(final Object[] val)
     {
         TypeEncoding<?> encoder = calculateEncoder(val,_encoder);
@@ -102,7 +104,6 @@
 
     private static TypeEncoding<?> calculateEncoder(final Object[] val, final EncoderImpl encoder)
     {
-
         if(val.length == 0)
         {
             AMQPType underlyingType = encoder.getTypeFromClass(val.getClass().getComponentType());
@@ -269,7 +270,6 @@
 
                     size +=  componentEncoding.getConstructorSize()
                                 + componentEncoding.getValueSize(null) * componentCount;
-
                 }
                 else
                 {
@@ -281,16 +281,19 @@
         return size;
     }
 
+    @Override
     public ArrayEncoding getCanonicalEncoding()
     {
         return _arrayEncoding;
     }
 
+    @Override
     public Collection<ArrayEncoding> getAllEncodings()
     {
         return Arrays.asList(_shortArrayEncoding, _arrayEncoding);
     }
 
+    @Override
     public void write(final Object[] val)
     {
         ArrayEncoding encoding = getEncoding(val);
@@ -444,25 +447,38 @@
             super(encoder, decoder);
         }
 
+        @Override
+        protected void writeSize(final Object[] val)
+        {
+            int encodedValueSize = getEncodedValueSize(val);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
+        }
+
+        @Override
         public void writeValue(final boolean[] a)
         {
             BooleanType.BooleanEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(boolean b : a)
             {
                 underlyingEncoder.writeValue(b);
             }
-
         }
 
+        @Override
         public void writeValue(final byte[] a)
         {
             ByteType.ByteEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(byte b : a)
@@ -471,11 +487,14 @@
             }
         }
 
+        @Override
         public void writeValue(final short[] a)
         {
             ShortType.ShortEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(short b : a)
@@ -484,12 +503,14 @@
             }
         }
 
+        @Override
         public void writeValue(final int[] a)
         {
-
             IntegerType.IntegerEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(int b : a)
@@ -498,12 +519,14 @@
             }
         }
 
+        @Override
         public void writeValue(final long[] a)
         {
-
             LongType.LongEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(long b : a)
@@ -512,12 +535,14 @@
             }
         }
 
+        @Override
         public void writeValue(final float[] a)
         {
-
             FloatType.FloatEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(float b : a)
@@ -526,12 +551,14 @@
             }
         }
 
+        @Override
         public void writeValue(final double[] a)
         {
-
             DoubleType.DoubleEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(double b : a)
@@ -540,12 +567,14 @@
             }
         }
 
+        @Override
         public void writeValue(final char[] a)
         {
-
             CharacterType.CharacterEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw(4 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null));
+            int encodedValueSize = 4 + underlyingEncoder.getConstructorSize() +
+                                   a.length * underlyingEncoder.getValueSize(null);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw(a.length);
             underlyingEncoder.writeConstructor();
             for(char b : a)
@@ -554,6 +583,7 @@
             }
         }
 
+        @Override
         public void setValue(final Object[] val, final TypeEncoding encoder, final int size)
         {
             _val = val;
@@ -602,16 +632,19 @@
             return EncodingCodes.ARRAY32;
         }
 
+        @Override
         public ArrayType getType()
         {
             return ArrayType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<Object[]> encoding)
         {
             return getType() == encoding.getType();
         }
 
+        @Override
         public Object[] readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -620,6 +653,7 @@
             return decodeArray(decoder, count);
         }
 
+        @Override
         public Object readValueArray()
         {
             DecoderImpl decoder = getDecoder();
@@ -628,6 +662,7 @@
             return decodeArrayAsObject(decoder, count);
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -637,13 +672,10 @@
         }
     }
 
-
-
     private class ShortArrayEncoding
             extends SmallFloatingSizePrimitiveTypeEncoding<Object[]>
             implements ArrayEncoding
     {
-
         private Object[] _val;
         private TypeEncoding _underlyingEncoder;
         private int _size;
@@ -653,26 +685,38 @@
             super(encoder, decoder);
         }
 
+        @Override
+        protected void writeSize(final Object[] val)
+        {
+            int encodedValueSize = getEncodedValueSize(val);
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw((byte) encodedValueSize);
+        }
+
+        @Override
         public void writeValue(final boolean[] a)
         {
             BooleanType.BooleanEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(boolean b : a)
             {
                 underlyingEncoder.writeValue(b);
             }
-
         }
 
+        @Override
         public void writeValue(final byte[] a)
         {
             ByteType.ByteEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
-            getEncoder().writeRaw((byte)a.length);
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             underlyingEncoder.writeConstructor();
             for(byte b : a)
             {
@@ -680,11 +724,14 @@
             }
         }
 
+        @Override
         public void writeValue(final short[] a)
         {
             ShortType.ShortEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(short b : a)
@@ -693,12 +740,14 @@
             }
         }
 
+        @Override
         public void writeValue(final int[] a)
         {
-
             IntegerType.IntegerEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(int b : a)
@@ -707,12 +756,14 @@
             }
         }
 
+        @Override
         public void writeValue(final long[] a)
         {
-
             LongType.LongEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(long b : a)
@@ -721,12 +772,14 @@
             }
         }
 
+        @Override
         public void writeValue(final float[] a)
         {
-
             FloatType.FloatEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(float b : a)
@@ -735,12 +788,14 @@
             }
         }
 
+        @Override
         public void writeValue(final double[] a)
         {
-
             DoubleType.DoubleEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(double b : a)
@@ -749,12 +804,14 @@
             }
         }
 
+        @Override
         public void writeValue(final char[] a)
         {
-
             CharacterType.CharacterEncoding underlyingEncoder = getUnderlyingEncoding(a);
-            getEncoder().writeRaw((byte)(1 + underlyingEncoder.getConstructorSize()
-                                  + a.length*underlyingEncoder.getValueSize(null)));
+            byte encodedValueSize = (byte)(1 + underlyingEncoder.getConstructorSize() +
+                                    a.length * underlyingEncoder.getValueSize(null));
+            getEncoder().getBuffer().ensureRemaining(encodedValueSize);
+            getEncoder().writeRaw(encodedValueSize);
             getEncoder().writeRaw((byte)a.length);
             underlyingEncoder.writeConstructor();
             for(char b : a)
@@ -763,6 +820,7 @@
             }
         }
 
+        @Override
         public void setValue(final Object[] val, final TypeEncoding encoder, final int size)
         {
             _val = val;
@@ -840,7 +898,6 @@
                     {
                         throw new IllegalArgumentException("Cannot encode arrays of type " + componentType.getName());
                     }
-
                 }
                 else
                 {
@@ -867,16 +924,19 @@
             return EncodingCodes.ARRAY8;
         }
 
+        @Override
         public ArrayType getType()
         {
             return ArrayType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<Object[]> encoding)
         {
             return getType() == encoding.getType();
         }
 
+        @Override
         public Object[] readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -885,6 +945,7 @@
             return decodeArray(decoder, count);
         }
 
+        @Override
         public Object readValueArray()
         {
             DecoderImpl decoder = getDecoder();
@@ -893,6 +954,7 @@
             return decodeArrayAsObject(decoder, count);
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -922,13 +984,11 @@
         }
     }
 
-
     private ByteType.ByteEncoding getUnderlyingEncoding(final byte[] a)
     {
         return _byteType.getCanonicalEncoding();
     }
 
-
     private ShortType.ShortEncoding getUnderlyingEncoding(final short[] a)
     {
         return _shortType.getCanonicalEncoding();
@@ -958,19 +1018,16 @@
         }
     }
 
-
     private FloatType.FloatEncoding getUnderlyingEncoding(final float[] a)
     {
         return _floatType.getCanonicalEncoding();
     }
 
-
     private DoubleType.DoubleEncoding getUnderlyingEncoding(final double[] a)
     {
         return _doubleType.getCanonicalEncoding();
     }
 
-
     private CharacterType.CharacterEncoding getUnderlyingEncoding(final char[] a)
     {
         return _characterType.getCanonicalEncoding();
@@ -1058,13 +1115,11 @@
             {
                 throw new ClassCastException("Unexpected class " + constructor.getClass().getName());
             }
-
         }
         else
         {
             return decodeNonPrimitive(decoder, constructor, count);
         }
-
     }
 
     private static boolean[] decodeBooleanArray(BooleanType.BooleanEncoding constructor, final int count)
@@ -1115,7 +1170,6 @@
         return array;
     }
 
-
     private static long[] decodeLongArray(LongType.LongEncoding constructor, final int count)
     {
         long[] array = new long[count];
@@ -1151,9 +1205,5 @@
 
         return array;
     }
-
-
-
-
 }
 
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 be0a8f5..c9f6d49 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
@@ -20,11 +20,11 @@
  */
 package org.apache.qpid.proton.codec;
 
-import org.apache.qpid.proton.amqp.Binary;
-
 import java.util.Arrays;
 import java.util.Collection;
 
+import org.apache.qpid.proton.amqp.Binary;
+
 public class BinaryType extends AbstractPrimitiveType<Binary>
 {
     private final BinaryEncoding _binaryEncoding;
@@ -43,21 +43,25 @@
         decoder.register(this);
     }
 
+    @Override
     public Class<Binary> getTypeClass()
     {
         return Binary.class;
     }
 
+    @Override
     public BinaryEncoding getEncoding(final Binary val)
     {
         return val.getLength() <= 255 ? _shortBinaryEncoding : _binaryEncoding;
     }
 
+    @Override
     public BinaryEncoding getCanonicalEncoding()
     {
         return _binaryEncoding;
     }
 
+    @Override
     public Collection<BinaryEncoding> getAllEncodings()
     {
         return Arrays.asList(_shortBinaryEncoding, _binaryEncoding);
@@ -67,12 +71,16 @@
     {
         if (binary.getLength() <= 255)
         {
+            // Reserve size of body + type encoding and single byte size
+            encoder.getBuffer().ensureRemaining(2 + binary.getLength());
             encoder.writeRaw(EncodingCodes.VBIN8);
             encoder.writeRaw((byte) binary.getLength());
             encoder.writeRaw(binary.getArray(), binary.getArrayOffset(), binary.getLength());
         }
         else
         {
+            // Reserve size of body + type encoding and four byte size
+            encoder.getBuffer().ensureRemaining(5 + binary.getLength());
             encoder.writeRaw(EncodingCodes.VBIN32);
             encoder.writeRaw(binary.getLength());
             encoder.writeRaw(binary.getArray(), binary.getArrayOffset(), binary.getLength());
@@ -92,6 +100,7 @@
         @Override
         protected void writeEncodedValue(final Binary val)
         {
+            getEncoder().getBuffer().ensureRemaining(val.getLength());
             getEncoder().writeRaw(val.getArray(), val.getArrayOffset(), val.getLength());
         }
 
@@ -101,23 +110,25 @@
             return val.getLength();
         }
 
-
         @Override
         public byte getEncodingCode()
         {
             return EncodingCodes.VBIN32;
         }
 
+        @Override
         public BinaryType getType()
         {
             return BinaryType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<Binary> encoding)
         {
             return (getType() == encoding.getType());
         }
 
+        @Override
         public Binary readValue()
         {
             final DecoderImpl decoder = getDecoder();
@@ -131,6 +142,7 @@
             return new Binary(data);
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -153,6 +165,7 @@
         @Override
         protected void writeEncodedValue(final Binary val)
         {
+            getEncoder().getBuffer().ensureRemaining(val.getLength());
             getEncoder().writeRaw(val.getArray(), val.getArrayOffset(), val.getLength());
         }
 
@@ -162,23 +175,25 @@
             return val.getLength();
         }
 
-
         @Override
         public byte getEncodingCode()
         {
             return EncodingCodes.VBIN8;
         }
 
+        @Override
         public BinaryType getType()
         {
             return BinaryType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<Binary> encoder)
         {
             return encoder == this;
         }
 
+        @Override
         public Binary readValue()
         {
             int size = ((int)getDecoder().readRawByte()) & 0xff;
@@ -187,6 +202,7 @@
             return new Binary(data);
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
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 a3ff4af..265063c 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
@@ -48,11 +48,13 @@
         decoder.register(this);
     }
 
+    @Override
     public Class<List> getTypeClass()
     {
         return List.class;
     }
 
+    @Override
     public ListEncoding getEncoding(final List val)
     {
         int calculatedSize = calculateSize(val, _encoder);
@@ -85,11 +87,13 @@
         return len;
     }
 
+    @Override
     public ListEncoding getCanonicalEncoding()
     {
         return _listEncoding;
     }
 
+    @Override
     public Collection<ListEncoding> getAllEncodings()
     {
         return Arrays.asList(_zeroListEncoding, _shortListEncoding, _listEncoding);
@@ -111,6 +115,7 @@
         @Override
         protected void writeEncodedValue(final List val)
         {
+            getEncoder().getBuffer().ensureRemaining(getSizeBytes() + getEncodedValueSize(val));
             getEncoder().writeRaw(val.size());
 
             final int count = val.size();
@@ -137,16 +142,19 @@
             return EncodingCodes.LIST32;
         }
 
+        @Override
         public ListType getType()
         {
             return ListType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<List> encoding)
         {
             return (getType() == encoding.getType());
         }
 
+        @Override
         public List readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -224,6 +232,7 @@
             return list;
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -232,6 +241,7 @@
             buffer.position(buffer.position() + size);
         }
 
+        @Override
         public void setValue(final List value, final int length)
         {
             _value = value;
@@ -255,6 +265,7 @@
         @Override
         protected void writeEncodedValue(final List val)
         {
+            getEncoder().getBuffer().ensureRemaining(getSizeBytes() + getEncodedValueSize(val));
             getEncoder().writeRaw((byte)val.size());
 
             final int count = val.size();
@@ -281,16 +292,19 @@
             return EncodingCodes.LIST8;
         }
 
+        @Override
         public ListType getType()
         {
             return ListType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<List> encoder)
         {
             return encoder == this;
         }
 
+        @Override
         public List readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -363,6 +377,7 @@
             return list;
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -371,6 +386,7 @@
             buffer.position(buffer.position() + size);
         }
 
+        @Override
         public void setValue(final List value, final int length)
         {
             _value = value;
@@ -400,24 +416,29 @@
             return 0;
         }
 
+        @Override
         public ListType getType()
         {
            return ListType.this;
         }
 
+        @Override
         public void setValue(List value, int length)
         {
         }
 
+        @Override
         public void writeValue(final List val)
         {
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<List> encoder)
         {
             return encoder == this;
         }
 
+        @Override
         public List readValue()
         {
             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 b0e2330..b791c91 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
@@ -153,6 +153,7 @@
         @Override
         protected void writeEncodedValue(final Map map)
         {
+            getEncoder().getBuffer().ensureRemaining(getSizeBytes() + getEncodedValueSize(map));
             getEncoder().writeRaw(2 * map.size());
 
             Iterator<Map.Entry> iter = map.entrySet().iterator();
@@ -260,6 +261,7 @@
             return map;
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -292,6 +294,7 @@
         @Override
         protected void writeEncodedValue(final Map map)
         {
+            getEncoder().getBuffer().ensureRemaining(getSizeBytes() + getEncodedValueSize(map));
             getEncoder().writeRaw((byte)(2 * map.size()));
 
             Iterator<Map.Entry> iter = map.entrySet().iterator();
@@ -394,6 +397,7 @@
             return map;
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
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 24bea47..7687ca4 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
@@ -30,6 +30,7 @@
     private static final DecoderImpl.TypeDecoder<String> _stringCreator =
         new DecoderImpl.TypeDecoder<String>()
         {
+            @Override
             public String decode(DecoderImpl decoder, final ReadableBuffer buffer)
             {
                 CharsetDecoder charsetDecoder = decoder.getCharsetDecoder();
@@ -64,11 +65,13 @@
         decoder.register(this);
     }
 
+    @Override
     public Class<String> getTypeClass()
     {
         return String.class;
     }
 
+    @Override
     public StringEncoding getEncoding(final String val)
     {
         final int length = calculateUTF8Length(val);
@@ -103,11 +106,13 @@
         return len;
     }
 
+    @Override
     public StringEncoding getCanonicalEncoding()
     {
         return _stringEncoding;
     }
 
+    @Override
     public Collection<StringEncoding> getAllEncodings()
     {
         return Arrays.asList(_shortStringEncoding, _stringEncoding);
@@ -128,6 +133,7 @@
         @Override
         protected void writeEncodedValue(final String val)
         {
+            getEncoder().getBuffer().ensureRemaining(getEncodedValueSize(val));
             getEncoder().writeRaw(val);
         }
 
@@ -143,16 +149,19 @@
             return EncodingCodes.STR32;
         }
 
+        @Override
         public StringType getType()
         {
             return StringType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<String> encoding)
         {
             return (getType() == encoding.getType());
         }
 
+        @Override
         public String readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -160,12 +169,14 @@
             return size == 0 ? "" : decoder.readRaw(_stringCreator, size);
         }
 
+        @Override
         public void setValue(final String val, final int length)
         {
             _value = val;
             _length = length;
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -190,6 +201,7 @@
         @Override
         protected void writeEncodedValue(final String val)
         {
+            getEncoder().getBuffer().ensureRemaining(getEncodedValueSize(val));
             getEncoder().writeRaw(val);
         }
 
@@ -205,16 +217,19 @@
             return EncodingCodes.STR8;
         }
 
+        @Override
         public StringType getType()
         {
             return StringType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<String> encoder)
         {
             return encoder == this;
         }
 
+        @Override
         public String readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -222,12 +237,14 @@
             return size == 0 ? "" : decoder.readRaw(_stringCreator, size);
         }
 
+        @Override
         public void setValue(final String val, final int length)
         {
             _value = val;
             _length = length;
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
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 00051ac..6c89cba 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
@@ -20,14 +20,14 @@
  */
 package org.apache.qpid.proton.codec;
 
-import org.apache.qpid.proton.amqp.Symbol;
-
 import java.nio.charset.Charset;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.qpid.proton.amqp.Symbol;
+
 public class SymbolType extends AbstractPrimitiveType<Symbol>
 {
     private static final Charset ASCII_CHARSET = Charset.forName("US-ASCII");
@@ -69,6 +69,7 @@
         decoder.register(this);
     }
 
+    @Override
     public Class<Symbol> getTypeClass()
     {
         return Symbol.class;
@@ -78,28 +79,35 @@
     {
         if (symbol.length() <= 255)
         {
+            // Reserve size of body + type encoding and single byte size
+            encoder.getBuffer().ensureRemaining(2 + symbol.length());
             encoder.writeRaw(EncodingCodes.SYM8);
             encoder.writeRaw((byte) symbol.length());
             symbol.writeTo(encoder.getBuffer());
         }
         else
         {
+            // Reserve size of body + type encoding and four byte size
+            encoder.getBuffer().ensureRemaining(5 + symbol.length());
             encoder.writeRaw(EncodingCodes.SYM32);
             encoder.writeRaw(symbol.length());
             symbol.writeTo(encoder.getBuffer());
         }
     }
 
+    @Override
     public SymbolEncoding getEncoding(final Symbol val)
     {
         return val.length() <= 255 ? _shortSymbolEncoding : _symbolEncoding;
     }
 
+    @Override
     public SymbolEncoding getCanonicalEncoding()
     {
         return _symbolEncoding;
     }
 
+    @Override
     public Collection<SymbolEncoding> getAllEncodings()
     {
         return Arrays.asList(_shortSymbolEncoding, _symbolEncoding);
@@ -118,6 +126,7 @@
         @Override
         protected void writeEncodedValue(final Symbol val)
         {
+            getEncoder().getBuffer().ensureRemaining(getEncodedValueSize(val));
             val.writeTo(getEncoder().getBuffer());
         }
 
@@ -127,23 +136,25 @@
             return val.length();
         }
 
-
         @Override
         public byte getEncodingCode()
         {
             return EncodingCodes.SYM32;
         }
 
+        @Override
         public SymbolType getType()
         {
             return SymbolType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<Symbol> encoding)
         {
             return (getType() == encoding.getType());
         }
 
+        @Override
         public Symbol readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -151,6 +162,7 @@
             return decoder.readRaw(_symbolCreator, size);
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -173,6 +185,7 @@
         @Override
         protected void writeEncodedValue(final Symbol val)
         {
+            getEncoder().getBuffer().ensureRemaining(getEncodedValueSize(val));
             val.writeTo(getEncoder().getBuffer());
         }
 
@@ -182,23 +195,25 @@
             return val.length();
         }
 
-
         @Override
         public byte getEncodingCode()
         {
             return EncodingCodes.SYM8;
         }
 
+        @Override
         public SymbolType getType()
         {
             return SymbolType.this;
         }
 
+        @Override
         public boolean encodesSuperset(final TypeEncoding<Symbol> encoder)
         {
             return encoder == this;
         }
 
+        @Override
         public Symbol readValue()
         {
             DecoderImpl decoder = getDecoder();
@@ -206,6 +221,7 @@
             return decoder.readRaw(_symbolCreator, size);
         }
 
+        @Override
         public void skipValue()
         {
             DecoderImpl decoder = getDecoder();
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/codec/WritableBuffer.java b/proton-j/src/main/java/org/apache/qpid/proton/codec/WritableBuffer.java
index 51e6079..034a23e 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/codec/WritableBuffer.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/codec/WritableBuffer.java
@@ -41,6 +41,10 @@
 
     boolean hasRemaining();
 
+    default void ensureRemaining(int requiredRemaining) {
+        // No-op to allow for drop in updates
+    }
+
     int remaining();
 
     int position();
@@ -135,6 +139,19 @@
         }
 
         @Override
+        public void ensureRemaining(int remaining) {
+            if (remaining < 0) {
+                throw new IllegalArgumentException("Required remaining bytes cannot be negative");
+            }
+
+            if (_buf.remaining() < remaining) {
+                throw new IndexOutOfBoundsException(String.format(
+                    "Requested min remaining bytes(%d) exceeds remaining(%d) in underlying ByteBuffer: %s",
+                    remaining, _buf.remaining(), _buf));
+            }
+        }
+
+        @Override
         public int remaining() {
             return _buf.remaining();
         }
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java
index bfda8ab..13d481c 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java
@@ -20,6 +20,9 @@
  */
 package org.apache.qpid.proton.engine.impl;
 
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+
 import org.apache.qpid.proton.amqp.Binary;
 import org.apache.qpid.proton.amqp.transport.EmptyFrame;
 import org.apache.qpid.proton.amqp.transport.FrameBody;
@@ -28,9 +31,6 @@
 import org.apache.qpid.proton.codec.WritableBuffer;
 import org.apache.qpid.proton.framing.TransportFrame;
 
-import java.nio.BufferOverflowException;
-import java.nio.ByteBuffer;
-
 /**
  * FrameWriter
  *
@@ -128,7 +128,7 @@
                     break;
                 }
             }
-            catch (BufferOverflowException e)
+            catch (BufferOverflowException | IndexOutOfBoundsException e)
             {
                 grow();
             }
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
index 6f11956..778b509 100644
--- 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
@@ -27,6 +27,7 @@
 
 import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class ApplicationPropertiesTypeTest extends CodecTestSupport {
 
@@ -86,4 +87,26 @@
             assertEquals(new Date(currentTime + 100), decoded.getValue().get("date-2"));
         }
     }
+
+    @Test
+    public void testEncodeApplicationPropertiesReservesSpaceForPayload() throws IOException {
+        final int ENTRIES = 8;
+
+        Map<String, Object> propertiesMap = new LinkedHashMap<>();
+        ApplicationProperties properties = new ApplicationProperties(propertiesMap);
+
+        for (int i = 0; i < ENTRIES; ++i) {
+            properties.getValue().put(String.valueOf(i), i);
+        }
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(this.buffer);
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeObject(properties);
+
+        // Check that the Type tries to reserve space, actual encoding size not computed here.
+        // Each key should also try and reserve space for the String data
+        Mockito.verify(spy, Mockito.times(ENTRIES + 1)).ensureRemaining(Mockito.anyInt());
+    }
 }
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
index 2411b7e..ea5045b 100644
--- 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
@@ -29,6 +29,7 @@
 
 import org.apache.qpid.proton.amqp.Symbol;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 /**
  * Test decoding of AMQP Array types
@@ -274,4 +275,30 @@
             assertEquals(source[i], map[i]);
         }
     }
+
+    @Test
+    public void testEncodeSmallBooleanArrayReservesSpaceForPayload() throws IOException {
+        doTestEncodeBooleanArrayTypeReservation(SMALL_ARRAY_SIZE);
+    }
+
+    @Test
+    public void testEncodeLargeBooleanArrayReservesSpaceForPayload() throws IOException {
+        doTestEncodeBooleanArrayTypeReservation(LARGE_ARRAY_SIZE);
+    }
+
+    private void doTestEncodeBooleanArrayTypeReservation(int size) throws IOException {
+        boolean[] source = new boolean[size];
+        for (int i = 0; i < size; ++i) {
+            source[i] = i % 2 == 0;
+        }
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(this.buffer);
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeArray(source);
+
+        // Check that the ArrayType tries to reserve space, actual encoding size not computed here.
+        Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
+    }
 }
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/BinaryTypeCodecTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/BinaryTypeCodecTest.java
new file mode 100644
index 0000000..c84b93d
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/BinaryTypeCodecTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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 org.apache.qpid.proton.amqp.Binary;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+/**
+ * Test for the Proton Binary type encoder / decoder
+ */
+public class BinaryTypeCodecTest extends CodecTestSupport {
+
+    private final int LARGE_SIZE = 1024;
+    private final int SMALL_SIZE = 32;
+
+    @Test
+    public void testEncodeSmallBinaryReservesSpaceForPayload() throws IOException {
+        doTestEncodeBinaryTypeReservation(SMALL_SIZE);
+    }
+
+    @Test
+    public void testEncodeLargeBinaryReservesSpaceForPayload() throws IOException {
+        doTestEncodeBinaryTypeReservation(LARGE_SIZE);
+    }
+
+    private void doTestEncodeBinaryTypeReservation(int size) throws IOException {
+        byte[] data = new byte[size];
+        for (int i = 0; i < size; ++i) {
+            data[i] = (byte) (i % 255);
+        }
+
+        Binary binary = new Binary(data);
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(this.buffer);
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeBinary(binary);
+
+        // Check that the BinaryType tries to reserve space, actual encoding size not computed here.
+        Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
+    }
+}
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
index 6b5ad5d..3e964c8 100644
--- 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
@@ -30,6 +30,7 @@
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.UnsignedInteger;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 /**
  * Test for the Proton List encoder / decoder
@@ -117,4 +118,30 @@
             assertEquals(list.size(), resultList.size());
         }
     }
+
+    @Test
+    public void testEncodeSmallListReservesSpaceForPayload() throws IOException {
+        doTestEncodeListTypeReservation(SMALL_SIZE);
+    }
+
+    @Test
+    public void testEncodeLargeListReservesSpaceForPayload() throws IOException {
+        doTestEncodeListTypeReservation(LARGE_SIZE);
+    }
+
+    private void doTestEncodeListTypeReservation(int size) throws IOException {
+        List<Integer> list = new ArrayList<>();
+        for (int i = 0; i < size; ++i) {
+            list.add(i);
+        }
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(this.buffer);
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeList(list);
+
+        // Check that the ListType tries to reserve space, actual encoding size not computed here.
+        Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
+    }
 }
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
index 32a0119..73b4a70 100644
--- 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
@@ -21,11 +21,13 @@
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 import org.apache.qpid.proton.amqp.Binary;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class MapTypeCodecTest extends CodecTestSupport {
 
@@ -95,4 +97,30 @@
             assertEquals(map.size(), resultMap.size());
         }
     }
+
+    @Test
+    public void testEncodeSmallMapReservesSpaceForPayload() throws IOException {
+        doTestEncodeMapTypeReservation(SMALL_SIZE);
+    }
+
+    @Test
+    public void doTestEncodeMapTypeReservation() throws IOException {
+        doTestEncodeMapTypeReservation(LARGE_SIZE);
+    }
+
+    private void doTestEncodeMapTypeReservation(int size) throws IOException {
+        Map<Integer, Integer> map = new HashMap<>();
+        for (int i = 0; i < size; ++i) {
+            map.put(i, i);
+        }
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(this.buffer);
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeMap(map);
+
+        // Check that the MapType tries to reserve space, actual encoding size not computed here.
+        Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
+    }
 }
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 03297f1..59e4661 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
@@ -23,6 +23,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import java.io.IOException;
 import java.lang.Character.UnicodeBlock;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
@@ -35,6 +36,7 @@
 
 import org.apache.qpid.proton.amqp.messaging.AmqpValue;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 /**
  * Test the encoding and decoding of {@link StringType} values.
@@ -249,6 +251,36 @@
     }
 
     @Test
+    public void testEncodeSmallStringReservesSpaceForPayload() throws IOException {
+        doTestEncodeStringTypeReservation(32);
+    }
+
+    @Test
+    public void testEncodeLargeStringReservesSpaceForPayload() throws IOException {
+        doTestEncodeStringTypeReservation(512);
+    }
+
+    private void doTestEncodeStringTypeReservation(int size) throws IOException {
+        final DecoderImpl decoder = new DecoderImpl();
+        final EncoderImpl encoder = new EncoderImpl(decoder);
+        AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+
+        StringBuilder builder = new StringBuilder(size);
+        for (int i = 0; i < size; ++i) {
+            builder.append(i);
+        }
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(ByteBuffer.allocate(2048));
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeString(builder.toString());
+
+        // Check that the StringType tries to reserve space, actual encoding size not computed here.
+        Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
+    }
+
+    @Test
     public void testEncodeAndDecodeUsingWritableBufferDefaultPutString() throws Exception {
         final DecoderImpl decoder = new DecoderImpl();
         final EncoderImpl encoder = new EncoderImpl(decoder);
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/SymbolTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/SymbolTypeTest.java
new file mode 100644
index 0000000..ebe8c87
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/SymbolTypeTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.util.Random;
+
+import org.apache.qpid.proton.amqp.Symbol;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+/**
+ * Test the encoding and decoding of {@link SymbolType} values.
+ */
+public class SymbolTypeTest extends CodecTestSupport {
+
+    private final int LARGE_SIZE = 512;
+    private final int SMALL_SIZE = 32;
+
+    @Test
+    public void testEncodeSmallSymbolReservesSpaceForPayload() throws IOException {
+        doTestEncodeSymbolTypeReservation(SMALL_SIZE);
+    }
+
+    @Test
+    public void testEncodeLargeSymbolReservesSpaceForPayload() throws IOException {
+        doTestEncodeSymbolTypeReservation(LARGE_SIZE);
+    }
+
+    private void doTestEncodeSymbolTypeReservation(int size) throws IOException {
+        Random random = new Random(System.currentTimeMillis());
+        StringBuilder builder = new StringBuilder(size);
+        for (int i = 0; i < size; ++i) {
+            builder.append((byte) random.nextInt(127));
+        }
+
+        Symbol symbol = Symbol.valueOf(builder.toString());
+
+        WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(ByteBuffer.allocate(2048));
+        WritableBuffer spy = Mockito.spy(writable);
+
+        encoder.setByteBuffer(spy);
+        encoder.writeSymbol(symbol);
+
+        // Check that the SymbolType tries to reserve space, actual encoding size not computed here.
+        Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
+    }
+}
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/WritableBufferTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/WritableBufferTest.java
new file mode 100644
index 0000000..b3900bb
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/WritableBufferTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+/**
+ * Tests for built in ByteBuffer wrapper of WritableBuffer
+ */
+public class WritableBufferTest {
+
+    @Test
+    public void testCreateAllocatedWrapper() {
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.allocate(10);
+
+        assertEquals(10, buffer.remaining());
+        assertEquals(0, buffer.position());
+        assertTrue(buffer.hasRemaining());
+    }
+
+    @Test
+    public void testCreateByteArrayWrapper() {
+        byte[] data = new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(10, buffer.remaining());
+        assertEquals(0, buffer.position());
+        assertTrue(buffer.hasRemaining());
+    }
+
+    @Test
+    public void testLimit() {
+        ByteBuffer data = ByteBuffer.allocate(100);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(data.capacity(), buffer.limit());
+    }
+
+    @Test
+    public void testRemaining() {
+        ByteBuffer data = ByteBuffer.allocate(100);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(data.limit(), buffer.remaining());
+        buffer.put((byte) 0);
+        assertEquals(data.limit() - 1, buffer.remaining());
+    }
+
+    @Test
+    public void testHasRemaining() {
+        ByteBuffer data = ByteBuffer.allocate(100);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertTrue(buffer.hasRemaining());
+        buffer.put((byte) 0);
+        assertTrue(buffer.hasRemaining());
+        data.position(data.limit());
+        assertFalse(buffer.hasRemaining());
+    }
+
+    @Test
+    public void testEnsureReminingThrowsWhenExpected() {
+        ByteBuffer data = ByteBuffer.allocate(100);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(data.capacity(), buffer.limit());
+        try {
+            buffer.ensureRemaining(1024);
+            fail("Should have thrown an error on request for more than is available.");
+        } catch (IndexOutOfBoundsException iobe) {}
+
+        try {
+            buffer.ensureRemaining(-1);
+            fail("Should have thrown an error on request for negative space.");
+        } catch (IllegalArgumentException iae) {}
+    }
+
+    @Test
+    public void testEnsureReminingDefaultImplementation() {
+        WritableBuffer buffer = new DefaultWritableBuffer();
+
+        try {
+            buffer.ensureRemaining(1024);
+        } catch (IndexOutOfBoundsException iobe) {
+            fail("Should not have thrown an error on request for more than is available.");
+        }
+
+        try {
+            buffer.ensureRemaining(-1);
+        } catch (IllegalArgumentException iae) {
+            fail("Should not have thrown an error on request for negative space.");
+        }
+    }
+
+    @Test
+    public void testGetPosition() {
+        ByteBuffer data = ByteBuffer.allocate(100);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(0, buffer.position());
+        data.put((byte) 0);
+        assertEquals(1, buffer.position());
+    }
+
+    @Test
+    public void testSetPosition() {
+        ByteBuffer data = ByteBuffer.allocate(100);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(0, data.position());
+        buffer.position(1);
+        assertEquals(1, data.position());
+    }
+
+    @Test
+    public void testPutByteBuffer() {
+        ByteBuffer input = ByteBuffer.allocate(1024);
+        input.put((byte) 1);
+        input.flip();
+
+        ByteBuffer data = ByteBuffer.allocate(1024);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(0, buffer.position());
+        buffer.put(input);
+        assertEquals(1, buffer.position());
+    }
+
+    @Test
+    public void testPutString() {
+        String ascii = new String("ASCII");
+
+        ByteBuffer data = ByteBuffer.allocate(1024);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(0, buffer.position());
+        buffer.put(ascii);
+        assertEquals(ascii.length(), buffer.position());
+    }
+
+    @Test
+    public void testPutReadableBuffer() {
+        doPutReadableBufferTestImpl(true);
+        doPutReadableBufferTestImpl(false);
+    }
+
+    private void doPutReadableBufferTestImpl(boolean readOnly) {
+        ByteBuffer buf = ByteBuffer.allocate(1024);
+        buf.put((byte) 1);
+        buf.flip();
+        if(readOnly) {
+            buf = buf.asReadOnlyBuffer();
+        }
+
+        ReadableBuffer input = new ReadableBuffer.ByteBufferReader(buf);
+
+        if(readOnly) {
+            assertFalse("Expected buffer not to hasArray()", input.hasArray());
+        } else {
+            assertTrue("Expected buffer to hasArray()", input.hasArray());
+        }
+
+        ByteBuffer data = ByteBuffer.allocate(1024);
+        WritableBuffer buffer = WritableBuffer.ByteBufferWrapper.wrap(data);
+
+        assertEquals(0, buffer.position());
+        buffer.put(input);
+        assertEquals(1, buffer.position());
+    }
+
+    //----- WritableBuffer implementation with no default overrides ----------//
+
+    private static class DefaultWritableBuffer implements WritableBuffer {
+
+        private final WritableBuffer backing;
+
+        public DefaultWritableBuffer() {
+            backing = WritableBuffer.ByteBufferWrapper.allocate(1024);
+        }
+
+        @Override
+        public void put(byte b) {
+            backing.put(b);
+        }
+
+        @Override
+        public void putFloat(float f) {
+            backing.putFloat(f);
+        }
+
+        @Override
+        public void putDouble(double d) {
+            backing.putDouble(d);
+        }
+
+        @Override
+        public void put(byte[] src, int offset, int length) {
+            backing.put(src, offset, length);
+        }
+
+        @Override
+        public void putShort(short s) {
+            backing.putShort(s);
+        }
+
+        @Override
+        public void putInt(int i) {
+            backing.putInt(i);
+        }
+
+        @Override
+        public void putLong(long l) {
+            backing.putLong(l);
+        }
+
+        @Override
+        public boolean hasRemaining() {
+            return backing.hasRemaining();
+        }
+
+        @Override
+        public int remaining() {
+            return backing.remaining();
+        }
+
+        @Override
+        public int position() {
+            return backing.position();
+        }
+
+        @Override
+        public void position(int position) {
+            backing.position(position);
+        }
+
+        @Override
+        public void put(ByteBuffer payload) {
+            backing.put(payload);
+        }
+
+        @Override
+        public void put(ReadableBuffer payload) {
+            backing.put(payload);
+        }
+
+        @Override
+        public int limit() {
+            return backing.limit();
+        }
+    }
+}