Handle null value in AMQP 1.0 codecs

Fixes #206

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/codec/QpidProtonCodec.java

@@ -96,6 +96,8 @@ private static Object convertApplicationProperty(Object value) {
       return ((Date) value).getTime();
     } else if (value instanceof Symbol) {
       return ((Symbol) value).toString();
+    } else if (value == null) {
+      return null;
     } else {
       throw new IllegalArgumentException(
           "Type not supported for an application property: " + value.getClass());
@@ -289,6 +291,8 @@ protected Object convertToQpidType(Object value) {
       return Symbol.getSymbol(value.toString());
     } else if (value instanceof byte[]) {
       return new Binary((byte[]) value);
+    } else if (value == null) {
+      return null;
     } else {
       throw new IllegalArgumentException(
           "Type not supported for an application property: " + value.getClass());

src/main/java/com/rabbitmq/stream/codec/SwiftMqCodec.java

@@ -66,6 +66,10 @@ private static Object convertAmqpMapValue(AMQPType value) {
       return ((AMQPUuid) value).getValue();
     } else if (value instanceof AMQPSymbol) {
       return ((AMQPSymbol) value).getValue();
+    } else if (value instanceof AMQPNull) {
+      return null;
+    } else if (value == null) {
+      return null;
     } else {
       throw new IllegalArgumentException(
           "Type not supported for an application property: " + value.getClass());
@@ -311,6 +315,8 @@ protected AMQPType convertToSwiftMqType(Object value) {
       return new AMQPSymbol(value.toString());
     } else if (value instanceof UUID) {
       return new AMQPUuid((UUID) value);
+    } else if (value == value) {
+      return AMQPNull.NULL;
     } else {
       throw new IllegalArgumentException(
           "Type not supported for an application property: " + value.getClass());

src/main/java/com/rabbitmq/stream/codec/SwiftMqMessageBuilder.java

@@ -301,19 +301,19 @@ protected void addEntryTimestamp(String key, long value) {
     }
 
     protected void addEntry(String key, UUID value) {
-      map.put(keyMaker.apply(key), new AMQPUuid(value));
+      map.put(keyMaker.apply(key), value == null ? AMQPNull.NULL : new AMQPUuid(value));
     }
 
     protected void addEntry(String key, byte[] value) {
-      map.put(keyMaker.apply(key), new AMQPBinary(value));
+      map.put(keyMaker.apply(key), value == null ? AMQPNull.NULL : new AMQPBinary(value));
     }
 
     protected void addEntry(String key, String value) {
-      map.put(keyMaker.apply(key), new AMQPString(value));
+      map.put(keyMaker.apply(key), value == null ? AMQPNull.NULL : new AMQPString(value));
     }
 
     protected void addEntrySymbol(String key, String value) {
-      map.put(keyMaker.apply(key), new AMQPSymbol(value));
+      map.put(keyMaker.apply(key), value == null ? AMQPNull.NULL : new AMQPSymbol(value));
     }
   }
 

src/test/java/com/rabbitmq/stream/codec/CodecsTest.java

@@ -212,6 +212,7 @@ void codecs(CodecCouple codecCouple) {
                   .entry("binary", binary)
                   .entry("string", string)
                   .entrySymbol("symbol", symbol)
+                  .entry("null", (String) null)
                   .messageBuilder()
                   .messageAnnotations()
                   .entry("annotations.boolean", Boolean.FALSE)
@@ -235,6 +236,7 @@ void codecs(CodecCouple codecCouple) {
                   .entry("annotations.binary", binary)
                   .entry("annotations.string", string)
                   .entrySymbol("annotations.symbol", symbol)
+                  .entry("annotations.null", (String) null)
                   .messageBuilder()
                   .build();
           Codec.EncodedMessage encoded = serializer.encode(outboundMessage);
@@ -356,6 +358,7 @@ void codecs(CodecCouple codecCouple) {
               .isNotNull()
               .isInstanceOf(String.class)
               .isEqualTo(symbol);
+          assertThat(inboundMessage.getApplicationProperties().get("null")).isNull();
 
           // message annotations
           assertThat(inboundMessage.getMessageAnnotations().get("annotations.boolean"))
@@ -453,6 +456,7 @@ void codecs(CodecCouple codecCouple) {
               .isNotNull()
               .isInstanceOf(String.class)
               .isEqualTo(symbol);
+          assertThat(inboundMessage.getMessageAnnotations().get("annotations.null")).isNull();
         });
   }