Added tests for errors in the source stream

This commit adds decoder/message-reader tests for errors in
the source data buffer publisher. Because the tests extend
AbstractDataBufferAllocatingTestCase, they also check whether
the buffers that precede the error in the stream are properly
released.

Issue: SPR-17025

spring-core/src/test/java/org/springframework/core/codec/ByteArrayDecoderTests.java

@@ -29,9 +29,7 @@ import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
-import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.*;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
 
 /**
  * @author Arjen Poutsma
@@ -67,6 +65,21 @@ public class ByteArrayDecoderTests extends AbstractDataBufferAllocatingTestCase
 				.verify();
 	}
 
+	@Test
+	public void decodeError() {
+		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> source =
+				Flux.just(fooBuffer).mergeWith(Flux.error(new RuntimeException()));
+		Flux<byte[]> output = this.decoder.decode(source,
+				ResolvableType.forClassWithGenerics(Publisher.class, byte[].class),
+				null, Collections.emptyMap());
+
+		StepVerifier.create(output)
+				.consumeNextWith(bytes -> assertArrayEquals("foo".getBytes(), bytes))
+				.expectError()
+				.verify();
+	}
+
 	@Test
 	public void decodeToMono() {
 		DataBuffer fooBuffer = stringBuffer("foo");

spring-core/src/test/java/org/springframework/core/codec/ByteBufferDecoderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -30,8 +30,7 @@ import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.*;
-import static org.junit.Assert.assertTrue;
 
 /**
  * @author Sebastien Deleuze
@@ -65,6 +64,21 @@ public class ByteBufferDecoderTests extends AbstractDataBufferAllocatingTestCase
 				.verify();
 	}
 
+	@Test
+	public void decodeError() {
+		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> source =
+				Flux.just(fooBuffer).mergeWith(Flux.error(new RuntimeException()));
+		Flux<ByteBuffer> output = this.decoder.decode(source,
+				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
+				null, Collections.emptyMap());
+
+		StepVerifier.create(output)
+				.expectNext(ByteBuffer.wrap("foo".getBytes()))
+				.expectError()
+				.verify();
+	}
+
 	@Test
 	public void decodeToMono() {
 		DataBuffer fooBuffer = stringBuffer("foo");

spring-core/src/test/java/org/springframework/core/codec/ResourceDecoderTests.java

@@ -32,7 +32,7 @@ import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StreamUtils;
 
 import static org.junit.Assert.*;
-import static org.springframework.core.ResolvableType.*;
+import static org.springframework.core.ResolvableType.forClass;
 
 /**
  * @author Arjen Poutsma
@@ -73,4 +73,19 @@ public class ResourceDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				.verify();
 	}
 
+	@Test
+	public void decodeError() {
+		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> source =
+				Flux.just(fooBuffer).mergeWith(Flux.error(new RuntimeException()));
+
+
+		Flux<Resource> result = this.decoder
+				.decode(source, forClass(Resource.class), null, Collections.emptyMap());
+
+		StepVerifier.create(result)
+				.expectError()
+				.verify();
+	}
+
 }

spring-core/src/test/java/org/springframework/core/codec/StringDecoderTests.java

@@ -173,6 +173,22 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	}
 
+	@Test
+	public void decodeError() {
+		DataBuffer fooBuffer = stringBuffer("foo\n");
+		Flux<DataBuffer> source =
+				Flux.just(fooBuffer).mergeWith(Flux.error(new RuntimeException()));
+
+		Flux<String> output = this.decoder.decode(source,
+				ResolvableType.forClass(String.class), null, Collections.emptyMap());
+
+		StepVerifier.create(output)
+				.expectNext("foo")
+				.expectError()
+				.verify();
+
+	}
+
 	@Test
 	public void decodeToMono() {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"), stringBuffer("baz"));

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -20,8 +20,14 @@ import java.util.List;
 import java.util.Map;
 
 import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
@@ -34,7 +40,7 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class FormHttpMessageReaderTests {
+public class FormHttpMessageReaderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private final FormHttpMessageReader reader = new FormHttpMessageReader();
 
@@ -96,8 +102,25 @@ public class FormHttpMessageReaderTests {
 		assertNull("Invalid result", result.getFirst("name 3"));
 	}
 
+	@Test
+	public void readFormError() {
+		DataBuffer fooBuffer = stringBuffer("name=value");
+		Flux<DataBuffer> body =
+				Flux.just(fooBuffer).mergeWith(Flux.error(new RuntimeException()));
+		MockServerHttpRequest request = request(body);
+
+		Flux<MultiValueMap<String, String>> result = this.reader.read(null, request, null);
+		StepVerifier.create(result)
+				.expectError()
+				.verify();
+	}
+
 
 	private MockServerHttpRequest request(String body) {
+		return request(Mono.just(stringBuffer(body)));
+	}
+
+	private MockServerHttpRequest request(Publisher<? extends DataBuffer> body) {
 		return MockServerHttpRequest
 					.method(HttpMethod.GET, "/")
 					.header(HttpHeaders.CONTENT_TYPE,  MediaType.APPLICATION_FORM_URLENCODED_VALUE)

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

@@ -21,10 +21,12 @@ import java.util.Collections;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
@@ -57,9 +59,10 @@ public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAll
 
 	@Test
 	public void readServerSentEvents() {
-		MockServerHttpRequest request = MockServerHttpRequest.post("/").body(
+		MockServerHttpRequest request = MockServerHttpRequest.post("/")
-				"id:c42\nevent:foo\nretry:123\n:bla\n:bla bla\n:bla bla bla\ndata:bar\n\n" +
+				.body(Mono.just(stringBuffer(
-				"id:c43\nevent:bar\nretry:456\ndata:baz\n\n");
+						"id:c42\nevent:foo\nretry:123\n:bla\n:bla bla\n:bla bla bla\ndata:bar\n\n" +
+						"id:c43\nevent:bar\nretry:456\ndata:baz\n\n")));
 
 		Flux<ServerSentEvent> events = this.messageReader
 				.read(ResolvableType.forClassWithGenerics(ServerSentEvent.class, String.class),
@@ -117,8 +120,9 @@ public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAll
 
 	@Test
 	public void readString() {
-		String body = "data:foo\ndata:bar\n\ndata:baz\n\n";
+
-		MockServerHttpRequest request = MockServerHttpRequest.post("/").body(body);
+		MockServerHttpRequest request = MockServerHttpRequest.post("/")
+				.body(Mono.just(stringBuffer("data:foo\ndata:bar\n\ndata:baz\n\n")));
 
 		Flux<String> data = messageReader.read(ResolvableType.forClass(String.class),
 				request, Collections.emptyMap()).cast(String.class);
@@ -132,9 +136,10 @@ public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAll
 
 	@Test
 	public void readPojo() {
-		MockServerHttpRequest request = MockServerHttpRequest.post("/").body(
+		MockServerHttpRequest request = MockServerHttpRequest.post("/")
-				"data:{\"foo\": \"foofoo\", \"bar\": \"barbar\"}\n\n" +
+				.body(Mono.just(stringBuffer(
-				"data:{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}\n\n");
+						"data:{\"foo\": \"foofoo\", \"bar\": \"barbar\"}\n\n" +
+								"data:{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}\n\n")));
 
 		Flux<Pojo> data = messageReader.read(ResolvableType.forClass(Pojo.class), request,
 				Collections.emptyMap()).cast(Pojo.class);
@@ -155,7 +160,8 @@ public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAll
 	@Test  // SPR-15331
 	public void decodeFullContentAsString() {
 		String body = "data:foo\ndata:bar\n\ndata:baz\n\n";
-		MockServerHttpRequest request = MockServerHttpRequest.post("/").body(body);
+		MockServerHttpRequest request = MockServerHttpRequest.post("/")
+				.body(Mono.just(stringBuffer(body)));
 
 		String actual = messageReader
 				.readMono(ResolvableType.forClass(String.class), request, Collections.emptyMap())
@@ -165,4 +171,25 @@ public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAll
 		assertEquals(body, actual);
 	}
 
+	@Test
+	public void readError() {
+
+		Flux<DataBuffer> body =
+				Flux.just(stringBuffer("data:foo\ndata:bar\n\ndata:baz\n\n"))
+						.mergeWith(Flux.error(new RuntimeException()));
+
+		MockServerHttpRequest request = MockServerHttpRequest.post("/")
+				.body(body);
+
+		Flux<String> data = messageReader.read(ResolvableType.forClass(String.class),
+				request, Collections.emptyMap()).cast(String.class);
+
+		StepVerifier.create(data)
+				.expectNextMatches(elem -> elem.equals("foo\nbar"))
+				.expectNextMatches(elem -> elem.equals("baz"))
+				.expectError()
+				.verify();
+	}
+
+
 }