Jackson2JsonDecoder tolerates null literal

Fixes #22042

spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -113,7 +113,7 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple
 				getObjectMapper().readerWithView(jsonView).forType(javaType) :
 				getObjectMapper().readerFor(javaType));
 
-		return tokens.map(tokenBuffer -> {
+		return tokens.flatMap(tokenBuffer -> {
 			try {
 				Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
 				if (!Hints.isLoggingSuppressed(hints)) {
@@ -122,16 +122,16 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple
 						return Hints.getLogPrefix(hints) + "Decoded [" + formatted + "]";
 					});
 				}
-				return value;
+				return Mono.justOrEmpty(value);
 			}
 			catch (InvalidDefinitionException ex) {
-				throw new CodecException("Type definition error: " + ex.getType(), ex);
+				return Mono.error(new CodecException("Type definition error: " + ex.getType(), ex));
 			}
 			catch (JsonProcessingException ex) {
-				throw new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex);
+				return Mono.error(new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex));
 			}
 			catch (IOException ex) {
-				throw new DecodingException("I/O error while parsing input stream", ex);
+				return Mono.error(new DecodingException("I/O error while parsing input stream", ex));
 			}
 		});
 	}

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -38,6 +38,7 @@ import org.springframework.core.codec.AbstractDecoderTestCase;
 import org.springframework.core.codec.CodecException;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
 import org.springframework.util.MimeType;
 
@@ -174,6 +175,14 @@ public class Jackson2JsonDecoderTests extends AbstractDecoderTestCase<Jackson2Js
 				.verifyError(DecodingException.class));
 	}
 
+	@Test // #22042
+	public void decodeWithNullLiteral() {
+		Flux<Object> result = this.decoder.decode(Flux.concat(stringBuffer("null")),
+				ResolvableType.forType(Pojo.class), MediaType.APPLICATION_JSON, Collections.emptyMap());
+
+		StepVerifier.create(result).expectComplete().verify();
+	}
+
 	@Test
 	public void noDefaultConstructor() {
 		Flux<DataBuffer> input =