Add error stream tests for Jackson2JsonDecoder

Issue: SPR-17418

Author: poutsma

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

@@ -46,10 +46,7 @@
 import static java.util.Arrays.asList;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 import static org.springframework.core.ResolvableType.forClass;
 import static org.springframework.http.MediaType.APPLICATION_JSON;
 import static org.springframework.http.MediaType.APPLICATION_JSON_UTF8;
@@ -229,6 +226,18 @@ public void invalidData() throws Exception {
 		StepVerifier.create(flux).verifyErrorMatches(ex -> ex instanceof DecodingException);
 	}
 
+	@Test
+	public void error() throws Exception {
+		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foofoo\": \"foofoo\", \"barbar\":"))
+				.concatWith(Flux.error(new RuntimeException()));
+		ResolvableType elementType = forClass(Pojo.class);
+		Flux<Object> flux = new Jackson2JsonDecoder(new ObjectMapper()).decode(source, elementType, null, emptyMap());
+
+		StepVerifier.create(flux)
+				.expectError(RuntimeException.class)
+				.verify();
+	}
+
 	@Test
 	public void noDefaultConstructor() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer( "{\"property1\":\"foo\",\"property2\":\"bar\"}"));

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

@@ -36,8 +36,8 @@
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 
-import static java.util.Arrays.*;
-import static java.util.Collections.*;
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
 
 /**
  * @author Arjen Poutsma
@@ -178,6 +178,19 @@ public void tokenizeArrayElements() {
 		testTokenize(asList("[1", ",2,", "3]"), asList("1", "2", "3"), true);
 	}
 
+	@Test
+	public void errorInStream() {
+		DataBuffer buffer = stringBuffer("{\"id\":1,\"name\":");
+		Flux<DataBuffer> source = Flux.just(buffer)
+				.concatWith(Flux.error(new RuntimeException()));
+
+		Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, true);
+
+		StepVerifier.create(result)
+				.expectError(RuntimeException.class)
+				.verify();
+	}
+
 	@Test(expected = DecodingException.class) // SPR-16521
 	public void jsonEOFExceptionIsWrappedAsDecodingError() {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"status\": \"noClosingQuote}"));