Update our tests to use TestSubscriber

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

@@ -17,8 +17,6 @@
 package org.springframework.core.codec.support;
 
 import java.nio.ByteBuffer;
-import java.util.List;
-import java.util.stream.StreamSupport;
 
 import org.junit.Test;
 import org.reactivestreams.Publisher;
@@ -28,8 +26,8 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -46,16 +44,14 @@ public class ByteBufferDecoderTests extends AbstractAllocatingTestCase {
 	}
 
 	@Test
-	public void decode() throws InterruptedException {
+	public void decode() {
 		DataBuffer fooBuffer = stringBuffer("foo");
 		DataBuffer barBuffer = stringBuffer("bar");
 		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
 		Flux<ByteBuffer> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null);
-		List<ByteBuffer> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(2, results.size());
-
-		assertBufferEquals(fooBuffer, results.get(0));
-		assertBufferEquals(barBuffer, results.get(1));
+		TestSubscriber<ByteBuffer> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValuesWith(b -> assertBufferEquals(fooBuffer, b), b -> assertBufferEquals(barBuffer, b));
 	}
 
 	public void assertBufferEquals(DataBuffer expected, ByteBuffer actual) {

spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java

@@ -18,8 +18,6 @@ package org.springframework.core.codec.support;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.stream.StreamSupport;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -30,8 +28,8 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -53,7 +51,7 @@ public class ByteBufferEncoderTests extends AbstractAllocatingTestCase {
 	}
 
 	@Test
-	public void encode() throws Exception {
+	public void encode() {
 		byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 		byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
 		Flux<ByteBuffer> source =
@@ -62,21 +60,17 @@ public class ByteBufferEncoderTests extends AbstractAllocatingTestCase {
 		Flux<DataBuffer> output = encoder.encode(source,
 				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
 				null);
-		List<DataBuffer> results =
-				StreamSupport.stream(output.toIterable().spliterator(), false)
-						.collect(toList());
-
-		assertEquals(2, results.size());
-		assertEquals(3, results.get(0).readableByteCount());
-		assertEquals(3, results.get(1).readableByteCount());
-
-		byte[] buf = new byte[3];
-		results.get(0).read(buf);
-		assertArrayEquals(fooBytes, buf);
-
-		results.get(1).read(buf);
-		assertArrayEquals(barBytes, buf);
-
+		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValuesWith(b -> {
+					byte[] buf = new byte[3];
+					b.read(buf);
+					assertArrayEquals(fooBytes, buf);
+				}, b -> {
+					byte[] buf = new byte[3];
+					b.read(buf);
+					assertArrayEquals(barBytes, buf);
+				});
 	}
 
 }

spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonDecoderTests.java

@@ -16,9 +16,6 @@
 
 package org.springframework.core.codec.support;
 
-import java.util.List;
-import java.util.stream.StreamSupport;
-
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
@@ -26,8 +23,8 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -43,13 +40,12 @@ public class JacksonJsonDecoderTests extends AbstractAllocatingTestCase {
 	}
 
 	@Test
-	public void decode() throws InterruptedException {
+	public void decode() {
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
 		Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
-		List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(1, results.size());
-		assertEquals("foofoo", ((Pojo) results.get(0)).getFoo());
+		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output).assertValues(new Pojo("foofoo", "barbar"));
 	}
 
 }

spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java

@@ -17,8 +17,6 @@
 package org.springframework.core.codec.support;
 
 import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.stream.StreamSupport;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -26,8 +24,8 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -48,17 +46,17 @@ public class JacksonJsonEncoderTests extends AbstractAllocatingTestCase {
 	}
 
 	@Test
-	public void write() throws InterruptedException {
+	public void write() {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
 		Flux<String> output = encoder.encode(source, null, null).map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
 		});
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(2, results.size());
-		assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", results.get(0));
-		assertEquals("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}", results.get(1));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}",
+							  "{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}");
 	}
 
 }

spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2DecoderTests.java

@@ -16,9 +16,6 @@
 
 package org.springframework.core.codec.support;
 
-import java.util.List;
-import java.util.stream.StreamSupport;
-
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
@@ -26,8 +23,8 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -44,13 +41,13 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 	}
 
 	@Test
-	public void decode() throws InterruptedException {
+	public void decode() {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(
 				"<?xml version=\"1.0\" encoding=\"UTF-8\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"));
 		Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
-		List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(1, results.size());
-		assertEquals("foofoo", ((Pojo) results.get(0)).getFoo());
+		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues(new Pojo("foofoo", "barbar"));
 	}
 
 }

spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java

@@ -17,8 +17,6 @@
 package org.springframework.core.codec.support;
 
 import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.stream.StreamSupport;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -26,8 +24,8 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -49,17 +47,17 @@ public class Jaxb2EncoderTests extends AbstractAllocatingTestCase {
 	}
 
 	@Test
-	public void encode() throws InterruptedException {
+	public void encode() {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
 		Flux<String> output = encoder.encode(source, null, null).map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
 		});
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(2, results.size());
-		assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>", results.get(0));
-		assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbarbar</bar><foo>foofoofoo</foo></pojo>", results.get(1));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>",
+							  "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbarbar</bar><foo>foofoofoo</foo></pojo>");
 	}
 
 }

spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectDecoderTests.java

@@ -17,16 +17,13 @@
 package org.springframework.core.codec.support;
 
 import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.stream.StreamSupport;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.io.buffer.DataBuffer;
 
-import static java.util.stream.Collectors.toList;
-import static org.junit.Assert.assertEquals;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -35,15 +32,15 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 
 
 	@Test
-	public void decodeSingleChunkToJsonObject() throws InterruptedException {
+	public void decodeSingleChunkToJsonObject()  {
 		JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
 		Flux<String> output =
 				decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
-		List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(1, results.size());
-		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}");
 	}
 
 	@Test
@@ -53,9 +50,9 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 				stringBuffer(", \"bar\": \"barbar\"}"));
 		Flux<String> output =
 				decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(1, results.size());
-		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}");
 	}
 
 	@Test
@@ -65,11 +62,10 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 				"[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
 		Flux<String> output =
 				decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
-
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(2, results.size());
-		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
-		assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}",
+							  "{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}");
 	}
 
 	@Test
@@ -80,10 +76,10 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 						": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
 		Flux<String> output =
 				decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(2, results.size());
-		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
-		assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}",
+							  "{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}");
 	}
 
 	private static String toString(DataBuffer buffer) {

spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java

@@ -25,7 +25,7 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.core.io.buffer.DataBuffer;
 
-import static org.junit.Assert.assertEquals;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -43,27 +43,28 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase {
 	public void encodeSingleElementFlux() throws InterruptedException {
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
-		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
+		Flux<String> output = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toIterable();
-		String result = String.join("", results);
-		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"}]", result);
+		});
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("[", "{\"foo\": \"foofoo\", \"bar\": \"barbar\"}]");
 	}
 
-
 	@Test
 	public void encodeSingleElementMono() throws InterruptedException {
 		Mono<DataBuffer> source =
 				Mono.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
-		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
+		Flux<String> output = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toIterable();
-		String result = String.join("", results);
-		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", result);
+		});
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}");
 	}
 
 	@Test
@@ -71,13 +72,16 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase {
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"),
 						stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}"));
-		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
+		Flux<String> output = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toIterable();
-		String result = String.join("", results);
-		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]", result);
+		});
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("[",
+						"{\"foo\": \"foofoo\", \"bar\": \"barbar\"},",
+						"{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]");
 	}
 
 	@Test
@@ -85,16 +89,19 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase {
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"),
 						stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}"),
-						stringBuffer(
-								"{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}")
+						stringBuffer("{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}")
 		);
-		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
+		Flux<String> output = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toIterable();
-		String result = String.join("", results);
-		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"},{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}]", result);
+		});
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output)
+				.assertValues("[",
+						"{\"foo\": \"foofoo\", \"bar\": \"barbar\"},",
+						"{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"},",
+						"{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}]");
 	}
 
 }

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

@@ -16,21 +16,18 @@
 
 package org.springframework.core.codec.support;
 
-import java.util.List;
-import java.util.stream.StreamSupport;
-
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.converter.RxJava1SingleConverter;
+import reactor.core.test.TestSubscriber;
 import rx.Single;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
 
 /**
@@ -58,9 +55,8 @@ public class StringDecoderTests extends AbstractAllocatingTestCase {
 	public void decode() throws InterruptedException {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"));
 		Flux<String> output = this.decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null);
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(1, results.size());
-		assertEquals("foobar", results.get(0));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output).assertValues("foobar");
 	}
 
 	@Test
@@ -68,10 +64,8 @@ public class StringDecoderTests extends AbstractAllocatingTestCase {
 		StringDecoder decoder = new StringDecoder(allocator, false);
 		Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"));
 		Flux<String> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null);
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(2, results.size());
-		assertEquals("foo", results.get(0));
-		assertEquals("bar", results.get(1));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output).assertValues("foo", "bar");
 	}
 
 	@Test
@@ -80,8 +74,8 @@ public class StringDecoderTests extends AbstractAllocatingTestCase {
 		Mono<String> mono = Mono.from(this.decoder.decode(source,
 				ResolvableType.forClassWithGenerics(Mono.class, String.class),
 				MediaType.TEXT_PLAIN));
-		String result = mono.get();
-		assertEquals("foobar", result);
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(mono).assertValues("foobar");
 	}
 
 	@Test

spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java

@@ -17,8 +17,6 @@
 package org.springframework.core.codec.support;
 
 import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.stream.StreamSupport;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -29,8 +27,8 @@ import reactor.core.publisher.Flux;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.MediaType;
 
-import static java.util.stream.Collectors.toList;
 import static org.junit.Assert.*;
+import reactor.core.test.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -59,9 +57,8 @@ public class StringEncoderTests extends AbstractAllocatingTestCase {
 			chunk.read(b);
 			return new String(b, StandardCharsets.UTF_8);
 		});
-		List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
-		assertEquals(1, results.size());
-		assertEquals("foo", results.get(0));
+		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output).assertValues("foo");
 	}
 
 }