Fix consuming null in DemuxingObjectDecoder

serde-support/src/main/java/io/micronaut/serde/support/deserializers/DemuxingObjectDecoder.java

@@ -107,19 +107,13 @@ private DemuxerState.Entry entryForValue() throws IOException {
     @Override
     protected Decoder delegate() throws IOException {
         DemuxerState.Entry entry = entryForValue();
-        Decoder delegate = entry.consume();
-        if (!consumeValues) {
-            JsonNode node = delegate.decodeNode();
-            entry.consumed = false;
-            entry.buffer = JsonNodeDecoder.create(node, LimitingStream.DEFAULT_LIMITS);
-            delegate = JsonNodeDecoder.create(node, LimitingStream.DEFAULT_LIMITS);
-        }
-        return delegate;
+        return entry.peekOrConsume(consumeValues);
     }
 
     @Override
     public boolean decodeNull() throws IOException {
-        return entryForValue().decodeNull();
+        DemuxerState.Entry entry = entryForValue();
+        return entry.peekOrConsumeNull(consumeValues);
     }
 
     @Override
@@ -213,29 +207,43 @@ private class Entry {
                 this.key = key;
             }
 
-            Decoder consume() {
-                if (consumed) {
-                    throw new IllegalStateException("Entry already consumed");
+            Decoder peekOrConsume(boolean consume) throws IOException {
+                Decoder decoder = provideDecoder(consume);
+                if (consume) {
+                    consumed = true;
                 }
-                consumed = true;
-                if (buffer != null) {
-                    return buffer;
-                } else {
-                    return delegate;
+                return decoder;
+            }
+
+            boolean peekOrConsumeNull(boolean consume) throws IOException {
+                Decoder decoder = provideDecoder(consume);
+                boolean isNull = decoder.decodeNull();
+                if (isNull && consume) {
+                    consumed = true;
                 }
+                return isNull;
             }
 
-            boolean decodeNull() throws IOException {
-                // this call has the expectation that a proper consume() will follow
+            private Decoder provideDecoder(boolean willConsume) throws IOException {
                 if (consumed) {
                     throw new IllegalStateException("Entry already consumed");
                 }
+                Decoder decoder;
                 if (buffer != null) {
-                    return buffer.decodeNull();
+                    decoder = buffer;
                 } else {
-                    return delegate.decodeNull();
+                    decoder = delegate;
+                }
+                if (willConsume) {
+                    return decoder;
+                } else {
+                    // if we don't consume it, we need to duplicate the data using a JsonNode.
+                    JsonNode node = decoder.decodeNode();
+                    buffer = JsonNodeDecoder.create(node, LimitingStream.DEFAULT_LIMITS);
+                    return JsonNodeDecoder.create(node, LimitingStream.DEFAULT_LIMITS);
                 }
             }
+
         }
     }
 

serde-support/src/test/groovy/io/micronaut/serde/support/deserializers/DemuxingObjectDecoderSpec.groovy

@@ -53,6 +53,7 @@ class DemuxingObjectDecoderSpec extends Specification {
         arr1.decodeInt() == 1
         arr1.finishStructure()
         demux1.decodeKey() == "b"
+        demux1.decodeNull() == false
         demux1.skipValue()
         demux1.decodeKey() == "c"
         def obj3 = demux1.decodeObject()
@@ -127,6 +128,72 @@ class DemuxingObjectDecoderSpec extends Specification {
         ctx.close()
     }
 
+    def 'simple structures with null'() {
+        given:
+            def ctx = ApplicationContext.run()
+            def outerDecoder = createDecoder(ctx, """{"a": [1], "b": null, "c": {"fizz": "buzz"}}""")
+
+            def primed = DemuxingObjectDecoder.prime(outerDecoder)
+            def demux1 = primed.decodeObject()
+            def demux2 = primed.decodeObject()
+
+        expect:
+            demux1.decodeKey() == "a"
+            def arr1 = demux1.decodeArray()
+            arr1.decodeInt() == 1
+            arr1.finishStructure()
+            demux1.decodeKey() == "b"
+            demux1.skipValue()
+            demux1.finishStructure(true)
+
+            demux2.decodeKey() == "b"
+            demux2.decodeNull()
+            demux2.decodeKey() == "c"
+            def obj3 = demux2.decodeObject()
+            obj3.decodeKey() == "fizz"
+            obj3.decodeString() == "buzz"
+            obj3.finishStructure()
+            demux2.decodeKey() == null
+            demux2.finishStructure()
+
+        cleanup:
+            ctx.close()
+    }
+
+    def 'simple structures with null not consuming'() {
+        given:
+            def ctx = ApplicationContext.run()
+            def outerDecoder = createDecoder(ctx, """{"a": [1], "b": null, "c": {"fizz": "buzz"}}""")
+
+            def primed = DemuxingObjectDecoder.prime(outerDecoder)
+            def demux1 = primed.decodeObjectNonConsuming()
+            def demux2 = primed.decodeObject()
+
+        expect:
+            demux1.decodeKey() == "a"
+            def arr1 = demux1.decodeArray()
+            arr1.decodeInt() == 1
+            arr1.finishStructure()
+            demux1.decodeKey() == "b"
+            demux1.decodeNull()
+            demux1.finishStructure(true)
+
+            demux2.decodeKey() == "a"
+            demux2.skipValue()
+            demux2.decodeKey() == "b"
+            demux2.decodeNull()
+            demux2.decodeKey() == "c"
+            def obj3 = demux2.decodeObject()
+            obj3.decodeKey() == "fizz"
+            obj3.decodeString() == "buzz"
+            obj3.finishStructure()
+            demux2.decodeKey() == null
+            demux2.finishStructure()
+
+        cleanup:
+            ctx.close()
+    }
+
     private static Decoder createDecoder(ApplicationContext ctx, @Language("json") String json) {
         JsonNodeDecoder.create(ctx.getBean(JsonMapper).readValue(json, JsonNode), LimitingStream.DEFAULT_LIMITS)
     }