Refactor tests with ScriptedSubscriber

Reactor recently added the `ScriptedSubscriber` in its new
`reactor-addons` module. This `Subscriber` revissits the previous
`TestSubscriber` with many improvements, including:

* scripting each expectation
* builder API that guides you until the final verification step
* virtual time support

This commit refactor all existing tests to use this new
infrastructure and removed the `TestSubscriber` implementation.

Issue: SPR-14800

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

@@ -28,6 +28,7 @@ import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.codec.ByteBufferDecoder;
 import org.springframework.core.codec.StringDecoder;
@@ -39,7 +40,7 @@ import org.springframework.http.ReactiveHttpInputMessage;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.http.codec.xml.Jaxb2XmlDecoder;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
-import org.springframework.tests.TestSubscriber;
+import org.springframework.web.server.UnsupportedMediaTypeStatusException;
 
 /**
  * @author Arjen Poutsma
@@ -79,9 +80,10 @@ public class BodyExtractorsTests {
 
 		Mono<String> result = extractor.extract(request, this.context);
 
-		TestSubscriber.subscribe(result)
-				.assertComplete()
-				.assertValues("foo");
+		ScriptedSubscriber.<String>create()
+				.expectNext("foo")
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
@@ -97,9 +99,11 @@ public class BodyExtractorsTests {
 		request.setBody(body);
 
 		Flux<String> result = extractor.extract(request, this.context);
-		TestSubscriber.subscribe(result)
-				.assertComplete()
-				.assertValues("foo");
+
+		ScriptedSubscriber.<String>create()
+				.expectNext("foo")
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
@@ -123,8 +127,9 @@ public class BodyExtractorsTests {
 		};
 
 		Flux<String> result = extractor.extract(request, emptyContext);
-		TestSubscriber.subscribe(result)
-				.assertError(UnsupportedMediaTypeException.class);
+		ScriptedSubscriber.create()
+				.expectError(UnsupportedMediaTypeException.class)
+				.verify(result);
 	}
 
 }

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

@@ -27,6 +27,7 @@ import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.codec.ByteBufferEncoder;
 import org.springframework.core.codec.CharSequenceEncoder;
@@ -39,7 +40,6 @@ import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
-import org.springframework.tests.TestSubscriber;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.assertArrayEquals;
@@ -79,14 +79,14 @@ public class BodyInsertersTests {
 
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		Mono<Void> result = inserter.insert(response, this.context);
-		TestSubscriber.subscribe(result)
-				.assertComplete();
+		ScriptedSubscriber.create().expectComplete().verify(result);
 
 		ByteBuffer byteBuffer = ByteBuffer.wrap(body.getBytes(UTF_8));
 		DataBuffer buffer = new DefaultDataBufferFactory().wrap(byteBuffer);
-		TestSubscriber.subscribe(response.getBody())
-				.assertComplete()
-				.assertValues(buffer);
+		ScriptedSubscriber.<DataBuffer>create()
+				.expectNext(buffer)
+				.expectComplete()
+				.verify(response.getBody());
 	}
 
 	@Test
@@ -98,14 +98,14 @@ public class BodyInsertersTests {
 
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		Mono<Void> result = inserter.insert(response, this.context);
-		TestSubscriber.subscribe(result)
-				.assertComplete();
+		ScriptedSubscriber.create().expectComplete().verify(result);
 
 		ByteBuffer byteBuffer = ByteBuffer.wrap("foo".getBytes(UTF_8));
 		DataBuffer buffer = new DefaultDataBufferFactory().wrap(byteBuffer);
-		TestSubscriber.subscribe(response.getBody())
-				.assertComplete()
-				.assertValues(buffer);
+		ScriptedSubscriber.<DataBuffer>create()
+				.expectNext(buffer)
+				.expectComplete()
+				.verify(response.getBody());
 	}
 
 	@Test
@@ -117,18 +117,18 @@ public class BodyInsertersTests {
 
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		Mono<Void> result = inserter.insert(response, this.context);
-		TestSubscriber.subscribe(result)
-				.assertComplete();
+		ScriptedSubscriber.create().expectComplete().verify(result);
 
 		byte[] expectedBytes = Files.readAllBytes(body.getFile().toPath());
 
-		TestSubscriber.subscribe(response.getBody())
-				.assertComplete()
-				.assertValuesWith(dataBuffer -> {
+		ScriptedSubscriber.<DataBuffer>create()
+				.consumeNextWith(dataBuffer -> {
 					byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
 					dataBuffer.read(resultBytes);
 					assertArrayEquals(expectedBytes, resultBytes);
-				});
+				})
+				.expectComplete()
+				.verify(response.getBody());
 	}
 
 	@Test
@@ -142,9 +142,7 @@ public class BodyInsertersTests {
 
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		Mono<Void> result = inserter.insert(response, this.context);
-		TestSubscriber.subscribe(result)
-				.assertComplete();
-
+		ScriptedSubscriber.create().expectNextCount(0).expectComplete().verify(result);
 	}
 
 	@Test
@@ -157,9 +155,7 @@ public class BodyInsertersTests {
 
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		Mono<Void> result = inserter.insert(response, this.context);
-		TestSubscriber.subscribe(result)
-				.assertComplete();
-
+		ScriptedSubscriber.create().expectNextCount(0).expectComplete().verify(result);
 	}
 
 }

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

@@ -15,26 +15,25 @@
  */
 package org.springframework.http.codec;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.junit.Assert.*;
-import static org.hamcrest.Matchers.*;
-
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 
 import org.jetbrains.annotations.NotNull;
 import org.junit.Test;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.ByteBufferEncoder;
 import org.springframework.core.codec.Encoder;
 import org.springframework.http.MediaType;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
-import org.springframework.tests.TestSubscriber;
 import org.springframework.util.MimeTypeUtils;
 
+import static java.nio.charset.StandardCharsets.*;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+
 /**
  * Unit tests for {@link EncoderHttpMessageWriter}.
  *
@@ -69,7 +68,10 @@ public class EncoderHttpMessageWriterTest {
 				MediaType.APPLICATION_OCTET_STREAM, this.response, Collections.emptyMap());
 
 		assertThat(this.response.getHeaders().getContentType(), is(MediaType.APPLICATION_OCTET_STREAM));
-		TestSubscriber.subscribe(this.response.getBodyAsString()).assertComplete().assertValues(payload);
+		ScriptedSubscriber.<String>create()
+				.expectNext(payload)
+				.expectComplete()
+				.verify(this.response.getBodyAsString());
 	}
 
 	@NotNull

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

@@ -22,6 +22,7 @@ import java.util.Collections;
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.ByteArrayResource;
@@ -32,7 +33,6 @@ import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
-import org.springframework.tests.TestSubscriber;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.hamcrest.Matchers.*;
@@ -69,23 +69,38 @@ public class ResourceHttpMessageWriterTests {
 
 	@Test
 	public void shouldWriteResource() throws Exception {
-		TestSubscriber.subscribe(this.writer.write(Mono.just(resource), null, ResolvableType.forClass(Resource.class),
-				MediaType.TEXT_PLAIN, this.request, this.response, Collections.emptyMap())).assertComplete();
+
+		Mono<Void> mono = this.writer.write(Mono.just(resource), null,
+				ResolvableType.forClass(Resource.class),
+				MediaType.TEXT_PLAIN, this.request, this.response, Collections.emptyMap());
+		ScriptedSubscriber
+				.<Void>create()
+				.expectNextCount(0)
+				.expectComplete()
+				.verify(mono);
 
 		assertThat(this.response.getHeaders().getContentType(), is(MediaType.TEXT_PLAIN));
 		assertThat(this.response.getHeaders().getContentLength(), is(39L));
 		assertThat(this.response.getHeaders().getFirst(HttpHeaders.ACCEPT_RANGES), is("bytes"));
 
 		Mono<String> result = this.response.getBodyAsString();
-		TestSubscriber.subscribe(result).assertComplete().assertValues("Spring Framework test resource content.");
+		ScriptedSubscriber
+				.<String>create()
+				.expectNext("Spring Framework test resource content.")
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
 	public void shouldWriteResourceRange() throws Exception {
 		this.request.getHeaders().setRange(Collections.singletonList(HttpRange.createByteRange(0, 5)));
-
-		TestSubscriber.subscribe(this.writer.write(Mono.just(resource), null, ResolvableType.forClass(Resource.class),
-				MediaType.TEXT_PLAIN, this.request, this.response, Collections.emptyMap())).assertComplete();
+		Mono<Void> mono = this.writer.write(Mono.just(resource), null, ResolvableType.forClass(Resource.class),
+				MediaType.TEXT_PLAIN, this.request, this.response, Collections.emptyMap());
+		ScriptedSubscriber
+				.<Void>create()
+				.expectNextCount(0)
+				.expectComplete()
+				.verify(mono);
 
 		assertThat(this.response.getHeaders().getContentType(), is(MediaType.TEXT_PLAIN));
 		assertThat(this.response.getHeaders().getFirst(HttpHeaders.CONTENT_RANGE), is("bytes 0-5/39"));
@@ -93,15 +108,24 @@ public class ResourceHttpMessageWriterTests {
 		assertThat(this.response.getHeaders().getContentLength(), is(6L));
 
 		Mono<String> result = this.response.getBodyAsString();
-		TestSubscriber.subscribe(result).assertComplete().assertValues("Spring");
+		ScriptedSubscriber
+				.<String>create()
+				.expectNext("Spring")
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
 	public void shouldSetRangeNotSatisfiableStatus() throws Exception {
 		this.request.getHeaders().set(HttpHeaders.RANGE, "invalid");
 
-		TestSubscriber.subscribe(this.writer.write(Mono.just(resource), null, ResolvableType.forClass(Resource.class),
-				MediaType.TEXT_PLAIN, this.request, this.response, Collections.emptyMap()));
+		Mono<Void> mono = this.writer.write(Mono.just(resource), null, ResolvableType.forClass(Resource.class),
+				MediaType.TEXT_PLAIN, this.request, this.response, Collections.emptyMap());
+		ScriptedSubscriber
+				.<Void>create()
+				.expectNextCount(0)
+				.expectComplete()
+				.verify(mono);
 
 		assertThat(this.response.getHeaders().getFirst(HttpHeaders.ACCEPT_RANGES), is("bytes"));
 		assertThat(this.response.getStatusCode(), is(HttpStatus.REQUESTED_RANGE_NOT_SATISFIABLE));

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

@@ -16,8 +16,6 @@
 
 package org.springframework.http.codec;
 
-import static org.junit.Assert.*;
-import static org.hamcrest.Matchers.*;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
@@ -30,6 +28,7 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.ByteArrayResource;
@@ -38,10 +37,12 @@ import org.springframework.core.io.support.ResourceRegion;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
-import org.springframework.tests.TestSubscriber;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StringUtils;
 
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+
 /**
  * Unit tests for {@link ResourceRegionHttpMessageWriter}.
  * @author Brian Clozel
@@ -82,15 +83,16 @@ public class ResourceRegionHttpMessageWriterTests {
 
 		ResourceRegion region = new ResourceRegion(this.resource, 0, 6);
 
-		TestSubscriber.subscribe(this.writer.write(Mono.just(region), ResolvableType.forClass(ResourceRegion.class),
-				MediaType.TEXT_PLAIN, this.response, Collections.emptyMap())).assertComplete();
+		Mono<Void> mono = this.writer.write(Mono.just(region), ResolvableType.forClass(ResourceRegion.class),
+				MediaType.TEXT_PLAIN, this.response, Collections.emptyMap());
+		ScriptedSubscriber.<Void>create().expectComplete().verify(mono);
 
 		assertThat(this.response.getHeaders().getContentType(), is(MediaType.TEXT_PLAIN));
 		assertThat(this.response.getHeaders().getFirst(HttpHeaders.CONTENT_RANGE), is("bytes 0-5/39"));
 		assertThat(this.response.getHeaders().getContentLength(), is(6L));
 
 		Mono<String> result = response.getBodyAsString();
-		TestSubscriber.subscribe(result).assertComplete().assertValues("Spring");
+		ScriptedSubscriber.<String>create().expectNext("Spring").expectComplete().verify(result);
 	}
 
 	@Test
@@ -105,43 +107,43 @@ public class ResourceRegionHttpMessageWriterTests {
 		Map<String, Object> hints = new HashMap<>(1);
 		hints.put(ResourceRegionHttpMessageWriter.BOUNDARY_STRING_HINT, boundary);
 
-		TestSubscriber.subscribe(
-				this.writer.write(regions, ResolvableType.forClass(ResourceRegion.class),
-						MediaType.TEXT_PLAIN, this.response, hints))
-				.assertComplete();
+		Mono<Void> mono = this.writer.write(regions, ResolvableType.forClass(ResourceRegion.class),
+				MediaType.TEXT_PLAIN, this.response, hints);
+		ScriptedSubscriber.<Void>create().expectComplete().verify(mono);
 
 		HttpHeaders headers = this.response.getHeaders();
 		assertThat(headers.getContentType().toString(), startsWith("multipart/byteranges;boundary=" + boundary));
 
 		Mono<String> result = response.getBodyAsString();
-		TestSubscriber
-				.subscribe(result).assertNoError()
-				.assertComplete()
-				.assertValuesWith(content -> {
-					String[] ranges = StringUtils.tokenizeToStringArray(content, "\r\n", false, true);
 
-					assertThat(ranges[0], is("--" + boundary));
-					assertThat(ranges[1], is("Content-Type: text/plain"));
-					assertThat(ranges[2], is("Content-Range: bytes 0-5/39"));
-					assertThat(ranges[3], is("Spring"));
-
-					assertThat(ranges[4], is("--" + boundary));
-					assertThat(ranges[5], is("Content-Type: text/plain"));
-					assertThat(ranges[6], is("Content-Range: bytes 7-15/39"));
-					assertThat(ranges[7], is("Framework"));
-
-					assertThat(ranges[8], is("--" + boundary));
-					assertThat(ranges[9], is("Content-Type: text/plain"));
-					assertThat(ranges[10], is("Content-Range: bytes 17-20/39"));
-					assertThat(ranges[11], is("test"));
-
-					assertThat(ranges[12], is("--" + boundary));
-					assertThat(ranges[13], is("Content-Type: text/plain"));
-					assertThat(ranges[14], is("Content-Range: bytes 22-38/39"));
-					assertThat(ranges[15], is("resource content."));
-
-					assertThat(ranges[16], is("--" + boundary + "--"));
-				});
+		ScriptedSubscriber
+				.<String>create()
+				.consumeNextWith(content -> {
+					String[] ranges = StringUtils
+							.tokenizeToStringArray(content, "\r\n", false, true);
+					String[] expected = new String[] {
+							"--" + boundary,
+							"Content-Type: text/plain",
+							"Content-Range: bytes 0-5/39",
+							"Spring",
+							"--" + boundary,
+							"Content-Type: text/plain",
+							"Content-Range: bytes 7-15/39",
+							"Framework",
+							"--" + boundary,
+							"Content-Type: text/plain",
+							"Content-Range: bytes 17-20/39",
+							"test",
+							"--" + boundary,
+							"Content-Type: text/plain",
+							"Content-Range: bytes 22-38/39",
+							"resource content.",
+							"--" + boundary + "--"
+					};
+					assertArrayEquals(expected, ranges);
+				})
+				.expectComplete()
+				.verify(result);
 	}
 
 }

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

@@ -18,11 +18,13 @@ package org.springframework.http.codec;
 
 import java.time.Duration;
 import java.util.Collections;
+import java.util.function.Consumer;
 
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -30,7 +32,6 @@ import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
-import org.springframework.tests.TestSubscriber;
 
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -71,17 +72,13 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
 		messageWriter.write(source, ResolvableType.forClass(ServerSentEvent.class),
 				new MediaType("text", "event-stream"), outputMessage, Collections.emptyMap());
 
-		Publisher<Publisher<DataBuffer>> result = outputMessage.getBodyWithFlush();
-		TestSubscriber.subscribe(result).
-				assertNoError().
-				assertValuesWith(publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError().assertValuesWith(
-							stringConsumer("id:c42\n" + "event:foo\n" + "retry:123\n" +
-									":bla\n:bla bla\n:bla bla bla\n" +
-									"data:bar\n"),
-							stringConsumer("\n"));
-
-				});
+		Publisher<Publisher<DataBuffer>> result = Flux.from(outputMessage.getBodyWithFlush());
+		ScriptedSubscriber
+				.<Publisher<DataBuffer>>create()
+				.consumeNextWith(sseConsumer("id:c42\n" + "event:foo\n" + "retry:123\n" +
+						":bla\n:bla bla\n:bla bla bla\n" + "data:bar\n"))
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
@@ -92,19 +89,12 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
 				new MediaType("text", "event-stream"), outputMessage, Collections.emptyMap());
 
 		Publisher<Publisher<DataBuffer>> result = outputMessage.getBodyWithFlush();
-		TestSubscriber.subscribe(result).
-				assertNoError().
-				assertValuesWith(publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError()
-							.assertValuesWith(stringConsumer("data:foo\n"),
-									stringConsumer("\n"));
-
-				}, publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError()
-							.assertValuesWith(stringConsumer("data:bar\n"),
-									stringConsumer("\n"));
-
-				});
+		ScriptedSubscriber
+				.<Publisher<DataBuffer>>create()
+				.consumeNextWith(sseConsumer("data:foo\n"))
+				.consumeNextWith(sseConsumer("data:bar\n"))
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
@@ -115,19 +105,12 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
 				new MediaType("text", "event-stream"), outputMessage, Collections.emptyMap());
 
 		Publisher<Publisher<DataBuffer>> result = outputMessage.getBodyWithFlush();
-		TestSubscriber.subscribe(result).
-				assertNoError().
-				assertValuesWith(publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError()
-							.assertValuesWith(stringConsumer("data:foo\ndata:bar\n"),
-									stringConsumer("\n"));
-
-				}, publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError()
-							.assertValuesWith(stringConsumer("data:foo\ndata:baz\n"),
-									stringConsumer("\n"));
-
-				});
+		ScriptedSubscriber
+				.<Publisher<DataBuffer>>create()
+				.consumeNextWith(sseConsumer("data:foo\ndata:bar\n"))
+				.consumeNextWith(sseConsumer("data:foo\ndata:baz\n"))
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
@@ -139,21 +122,22 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
 				new MediaType("text", "event-stream"), outputMessage, Collections.emptyMap());
 
 		Publisher<Publisher<DataBuffer>> result = outputMessage.getBodyWithFlush();
-		TestSubscriber.subscribe(result).
-				assertNoError().
-				assertValuesWith(publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError()
-							.assertValuesWith(stringConsumer("data:"), stringConsumer(
-									"{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"),
-									stringConsumer("\n"), stringConsumer("\n"));
+		ScriptedSubscriber
+				.<Publisher<DataBuffer>>create()
+				.consumeNextWith(sseConsumer("data:", "{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", "\n"))
+				.consumeNextWith(sseConsumer("data:", "{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}", "\n"))
+				.expectComplete()
+				.verify(result);
+	}
 
-				}, publisher -> {
-					TestSubscriber.subscribe(publisher).assertNoError()
-							.assertValuesWith(stringConsumer("data:"), stringConsumer(
-									"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"),
-									stringConsumer("\n"), stringConsumer("\n"));
-
-				});
+	private Consumer<Publisher<DataBuffer>> sseConsumer(String... expected) {
+		return publisher -> {
+			ScriptedSubscriber.StepBuilder<DataBuffer> builder = ScriptedSubscriber.create();
+			for (String value : expected) {
+				builder = builder.consumeNextWith(stringConsumer(value));
+			}
+			builder.consumeNextWith(stringConsumer("\n")).expectComplete().verify(publisher);
+		};
 	}
 
 }

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

@@ -25,15 +25,17 @@ import com.fasterxml.jackson.annotation.JsonView;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 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.Pojo;
-import org.springframework.tests.TestSubscriber;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Unit tests for {@link Jackson2JsonDecoder}.
@@ -53,18 +55,20 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 	}
 
 	@Test
-	public void decodePojo() {
+	public void decodePojo() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
 		ResolvableType elementType = ResolvableType.forClass(Pojo.class);
 		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null,
 				Collections.emptyMap());
 
-		TestSubscriber.subscribe(flux).assertNoError().assertComplete().
-				assertValues(new Pojo("foofoo", "barbar"));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(new Pojo("foofoo", "barbar"))
+				.expectComplete()
+				.verify(flux);
 	}
 
 	@Test
-	public void decodeToList() {
+	public void decodeToList() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(
 				"[{\"bar\":\"b1\",\"foo\":\"f1\"},{\"bar\":\"b2\",\"foo\":\"f2\"}]"));
 
@@ -72,12 +76,14 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 		Mono<Object> mono = new Jackson2JsonDecoder().decodeToMono(source, elementType,
 				null, Collections.emptyMap());
 
-		TestSubscriber.subscribe(mono).assertNoError().assertComplete().
-				assertValues(Arrays.asList(new Pojo("f1", "b1"), new Pojo("f2", "b2")));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(Arrays.asList(new Pojo("f1", "b1"), new Pojo("f2", "b2")))
+				.expectComplete()
+				.verify(mono);
 	}
 
 	@Test
-	public void decodeToFlux() {
+	public void decodeToFlux() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(
 				"[{\"bar\":\"b1\",\"foo\":\"f1\"},{\"bar\":\"b2\",\"foo\":\"f2\"}]"));
 
@@ -85,41 +91,43 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null,
 				Collections.emptyMap());
 
-		TestSubscriber.subscribe(flux).assertNoError().assertComplete().
-				assertValues(new Pojo("f1", "b1"), new Pojo("f2", "b2"));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(new Pojo("f1", "b1"))
+				.expectNext(new Pojo("f2", "b2"))
+				.expectComplete()
+				.verify(flux);
 	}
 
 	@Test
-	public void jsonView() {
+	public void jsonView() throws Exception {
 		Flux<DataBuffer> source = Flux.just(
 				stringBuffer("{\"withView1\" : \"with\", \"withView2\" : \"with\", \"withoutView\" : \"without\"}"));
-		ResolvableType elementType =  ResolvableType.forClass(JacksonViewBean.class);
+		ResolvableType elementType = ResolvableType.forClass(JacksonViewBean.class);
 		Map<String, Object> hints = Collections.singletonMap(Jackson2JsonDecoder.JSON_VIEW_HINT, MyJacksonView1.class);
 		Flux<JacksonViewBean> flux = new Jackson2JsonDecoder()
 				.decode(source, elementType, null, hints).cast(JacksonViewBean.class);
 
-		TestSubscriber
-				.subscribe(flux)
-				.assertNoError()
-				.assertComplete()
-				.assertValuesWith(b -> {
+		ScriptedSubscriber.<JacksonViewBean>create()
+				.consumeNextWith(b -> {
 					assertTrue(b.getWithView1().equals("with"));
 					assertNull(b.getWithView2());
 					assertNull(b.getWithoutView());
-				});
+				})
+				.expectComplete()
+				.verify(flux);
 	}
 
 	@Test
-	public void decodeEmptyBodyToMono() {
+	public void decodeEmptyBodyToMono() throws Exception {
 		Flux<DataBuffer> source = Flux.empty();
 		ResolvableType elementType = ResolvableType.forClass(Pojo.class);
-		Mono<Object> flux = new Jackson2JsonDecoder().decodeToMono(source, elementType,
+		Mono<Object> mono = new Jackson2JsonDecoder().decodeToMono(source, elementType,
 				null, Collections.emptyMap());
 
-		TestSubscriber.subscribe(flux)
-				.assertNoError()
-				.assertComplete()
-				.assertValueCount(0);
+		ScriptedSubscriber.
+				<Object>create().expectNextCount(0)
+				.expectComplete()
+				.verify(mono);
 	}
 
 

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

@@ -25,6 +25,7 @@ import com.fasterxml.jackson.annotation.JsonView;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -32,9 +33,9 @@ import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
 import org.springframework.http.codec.ServerSentEvent;
-import org.springframework.tests.TestSubscriber;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author Sebastien Deleuze
@@ -55,7 +56,7 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 	}
 
 	@Test
-	public void encode() {
+	public void encode() throws Exception {
 		Flux<Pojo> source = Flux.just(
 				new Pojo("foo", "bar"),
 				new Pojo("foofoo", "barbar"),
@@ -64,51 +65,52 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 		ResolvableType type = ResolvableType.forClass(Pojo.class);
 		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null, Collections.emptyMap());
 
-		TestSubscriber.subscribe(output)
-				.assertComplete()
-				.assertNoError()
-				.assertValuesWith(
-						stringConsumer("["),
-						stringConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}"),
-						stringConsumer(","),
-						stringConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"),
-						stringConsumer(","),
-						stringConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"),
-						stringConsumer("]")
-				);
+		ScriptedSubscriber
+				.<DataBuffer>create()
+				.consumeNextWith(stringConsumer("["))
+				.consumeNextWith(stringConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}"))
+				.consumeNextWith(stringConsumer(","))
+				.consumeNextWith(stringConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"))
+				.consumeNextWith(stringConsumer(","))
+				.consumeNextWith(stringConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"))
+				.consumeNextWith(stringConsumer("]"))
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
-	public void encodeWithType() {
+	public void encodeWithType() throws Exception {
 		Flux<ParentClass> source = Flux.just(new Foo(), new Bar());
 		ResolvableType type = ResolvableType.forClass(ParentClass.class);
 		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null, Collections.emptyMap());
 
-		TestSubscriber.subscribe(output)
-				.assertComplete()
-				.assertNoError()
-				.assertValuesWith(stringConsumer("["),
-						stringConsumer("{\"type\":\"foo\"}"),
-						stringConsumer(","),
-						stringConsumer("{\"type\":\"bar\"}"),
-						stringConsumer("]"));
+		ScriptedSubscriber
+				.<DataBuffer>create()
+				.consumeNextWith(stringConsumer("["))
+				.consumeNextWith(stringConsumer("{\"type\":\"foo\"}"))
+				.consumeNextWith(stringConsumer(","))
+				.consumeNextWith(stringConsumer("{\"type\":\"bar\"}"))
+				.consumeNextWith(stringConsumer("]"))
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
-	public void jsonView() {
+	public void jsonView() throws Exception {
 		JacksonViewBean bean = new JacksonViewBean();
 		bean.setWithView1("with");
 		bean.setWithView2("with");
 		bean.setWithoutView("without");
 
-		ResolvableType type =  ResolvableType.forClass(JacksonViewBean.class);
+		ResolvableType type = ResolvableType.forClass(JacksonViewBean.class);
 		Map<String, Object> hints = Collections.singletonMap(Jackson2JsonEncoder.JSON_VIEW_HINT, MyJacksonView1.class);
 		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);
 
-		TestSubscriber.subscribe(output)
-				.assertComplete()
-				.assertNoError()
-				.assertValuesWith(stringConsumer("{\"withView1\":\"with\"}"));
+		ScriptedSubscriber
+				.<DataBuffer>create()
+				.consumeNextWith(stringConsumer("{\"withView1\":\"with\"}"))
+				.expectComplete()
+				.verify(output);
 	}
 
 

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

@@ -21,10 +21,10 @@ import java.util.Collections;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.tests.TestSubscriber;
 
 /**
  * @author Sebastien Deleuze
@@ -32,15 +32,16 @@ import org.springframework.tests.TestSubscriber;
 public class JsonObjectDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
-	public void decodeSingleChunkToJsonObject()  {
+	public void decodeSingleChunkToJsonObject() throws Exception {
 		JsonObjectDecoder decoder = new JsonObjectDecoder();
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
 		Flux<String> output =
 				decoder.decode(source, null, null, Collections.emptyMap()).map(JsonObjectDecoderTests::toString);
-		TestSubscriber
-				.subscribe(output)
-				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}");
+		ScriptedSubscriber.<String>create()
+				.expectNext("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
@@ -50,9 +51,10 @@ public class JsonObjectDecoderTests extends AbstractDataBufferAllocatingTestCase
 				stringBuffer(", \"bar\": \"barbar\"}"));
 		Flux<String> output =
 				decoder.decode(source, null, null, Collections.emptyMap()).map(JsonObjectDecoderTests::toString);
-		TestSubscriber
-				.subscribe(output)
-				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}");
+		ScriptedSubscriber.<String>create()
+				.expectNext("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
@@ -62,10 +64,11 @@ public class JsonObjectDecoderTests extends AbstractDataBufferAllocatingTestCase
 				"[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
 		Flux<String> output =
 				decoder.decode(source, null, null, Collections.emptyMap()).map(JsonObjectDecoderTests::toString);
-		TestSubscriber
-				.subscribe(output)
-				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}",
-							  "{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}");
+		ScriptedSubscriber.<String>create()
+				.expectNext("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")
+				.expectNext("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}")
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
@@ -76,10 +79,11 @@ public class JsonObjectDecoderTests extends AbstractDataBufferAllocatingTestCase
 						": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
 		Flux<String> output =
 				decoder.decode(source, null, null, Collections.emptyMap()).map(JsonObjectDecoderTests::toString);
-		TestSubscriber
-				.subscribe(output)
-				.assertValues("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}",
-							  "{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}");
+		ScriptedSubscriber.<String>create()
+				.expectNext("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")
+				.expectNext("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}")
+				.expectComplete()
+				.verify(output);
 	}
 
 

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

@@ -23,6 +23,7 @@ import javax.xml.stream.events.XMLEvent;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -35,9 +36,10 @@ import org.springframework.http.codec.xml.jaxb.XmlRootElementWithNameAndNamespac
 import org.springframework.http.codec.xml.jaxb.XmlType;
 import org.springframework.http.codec.xml.jaxb.XmlTypeWithName;
 import org.springframework.http.codec.xml.jaxb.XmlTypeWithNameAndNamespace;
-import org.springframework.tests.TestSubscriber;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author Sebastien Deleuze
@@ -89,11 +91,8 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				.decode(Flux.just(stringBuffer(POJO_ROOT)), null, null, Collections.emptyMap());
 		Flux<List<XMLEvent>> result = this.decoder.split(xmlEvents, new QName("pojo"));
 
-		TestSubscriber
-				.subscribe(result)
-				.assertNoError()
-				.assertComplete()
-				.assertValuesWith(events -> {
+		ScriptedSubscriber.<List<XMLEvent>>create()
+				.consumeNextWith(events -> {
 					assertEquals(8, events.size());
 					assertStartElement(events.get(0), "pojo");
 					assertStartElement(events.get(1), "foo");
@@ -103,22 +102,20 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 					assertCharacters(events.get(5), "barbar");
 					assertEndElement(events.get(6), "bar");
 					assertEndElement(events.get(7), "pojo");
-				});
-
-
+				})
+				.expectComplete()
+				.verify(result);
 	}
 
 	@Test
-	public void splitMultipleBranches() {
+	public void splitMultipleBranches() throws Exception {
 		Flux<XMLEvent> xmlEvents = this.xmlEventDecoder
 				.decode(Flux.just(stringBuffer(POJO_CHILD)), null, null, Collections.emptyMap());
 		Flux<List<XMLEvent>> result = this.decoder.split(xmlEvents, new QName("pojo"));
 
-		TestSubscriber
-				.subscribe(result)
-				.assertNoError()
-				.assertComplete()
-				.assertValuesWith(events -> {
+
+		ScriptedSubscriber.<List<XMLEvent>>create()
+				.consumeNextWith(events -> {
 					assertEquals(8, events.size());
 					assertStartElement(events.get(0), "pojo");
 					assertStartElement(events.get(1), "foo");
@@ -128,7 +125,8 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 					assertCharacters(events.get(5), "bar");
 					assertEndElement(events.get(6), "bar");
 					assertEndElement(events.get(7), "pojo");
-				}, events -> {
+				})
+				.consumeNextWith(events -> {
 					assertEquals(8, events.size());
 					assertStartElement(events.get(0), "pojo");
 					assertStartElement(events.get(1), "foo");
@@ -138,7 +136,9 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 					assertCharacters(events.get(5), "barbar");
 					assertEndElement(events.get(6), "bar");
 					assertEndElement(events.get(7), "pojo");
-				});
+				})
+				.expectComplete()
+				.verify(result);
 	}
 
 	private static void assertStartElement(XMLEvent event, String expectedLocalName) {
@@ -162,11 +162,10 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		Flux<Object> output = this.decoder.decode(source, ResolvableType.forClass(Pojo.class),
 				null, Collections.emptyMap());
 
-		TestSubscriber
-				.subscribe(output)
-				.assertNoError()
-				.assertComplete()
-				.assertValues(new Pojo("foofoo", "barbar"));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(new Pojo("foofoo", "barbar"))
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
@@ -175,11 +174,10 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		Flux<Object> output = this.decoder.decode(source, ResolvableType.forClass(TypePojo.class),
 				null, Collections.emptyMap());
 
-		TestSubscriber
-				.subscribe(output)
-				.assertNoError()
-				.assertComplete()
-				.assertValues(new TypePojo("foofoo", "barbar"));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(new TypePojo("foofoo", "barbar"))
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
@@ -188,11 +186,11 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		Flux<Object> output = this.decoder.decode(source, ResolvableType.forClass(Pojo.class),
 				null, Collections.emptyMap());
 
-		TestSubscriber
-				.subscribe(output)
-				.assertNoError()
-				.assertComplete()
-				.assertValues(new Pojo("foo", "bar"), new Pojo("foofoo", "barbar"));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(new Pojo("foo", "bar"))
+				.expectNext(new Pojo("foofoo", "barbar"))
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test
@@ -201,11 +199,11 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		Flux<Object> output = this.decoder.decode(source, ResolvableType.forClass(TypePojo.class),
 				null, Collections.emptyMap());
 
-		TestSubscriber
-				.subscribe(output)
-				.assertNoError()
-				.assertComplete()
-				.assertValues(new TypePojo("foo", "bar"), new TypePojo("foofoo", "barbar"));
+		ScriptedSubscriber.<Object>create()
+				.expectNext(new TypePojo("foo", "bar"))
+				.expectNext(new TypePojo("foofoo", "barbar"))
+				.expectComplete()
+				.verify(output);
 	}
 
 	@Test

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

@@ -21,6 +21,7 @@ import java.util.Collections;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -29,10 +30,11 @@ import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
-import org.springframework.tests.TestSubscriber;
 
-import static org.junit.Assert.*;
-import static org.xmlunit.matchers.CompareMatcher.*;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.xmlunit.matchers.CompareMatcher.isSimilarTo;
 
 /**
  * @author Sebastien Deleuze
@@ -61,24 +63,27 @@ public class Jaxb2XmlEncoderTests extends AbstractDataBufferAllocatingTestCase {
 	}
 
 	@Test
-	public void encode() {
+	public void encode() throws Exception {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
 		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
 				ResolvableType.forClass(Pojo.class),
-						MediaType.APPLICATION_XML, Collections.emptyMap());
-		TestSubscriber
-				.subscribe(output)
-				.assertValuesWith(dataBuffer -> {
-			try {
-				String s = DataBufferTestUtils
-						.dumpString(dataBuffer, StandardCharsets.UTF_8);
-				assertThat(s, isSimilarTo("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
-						"<pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"));
-			}
-			finally {
-				DataBufferUtils.release(dataBuffer);
-			}
-		});
+				MediaType.APPLICATION_XML, Collections.emptyMap());
+
+		ScriptedSubscriber
+				.<DataBuffer>create()
+				.consumeNextWith(dataBuffer -> {
+					try {
+						String s = DataBufferTestUtils
+								.dumpString(dataBuffer, StandardCharsets.UTF_8);
+						assertThat(s, isSimilarTo("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
+								"<pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"));
+					}
+					finally {
+						DataBufferUtils.release(dataBuffer);
+					}
+				})
+				.expectComplete()
+				.verify(output);
 	}
 
 }

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

@@ -21,9 +21,9 @@ import javax.xml.stream.events.XMLEvent;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
-import org.springframework.tests.TestSubscriber;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -47,19 +47,19 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		Flux<XMLEvent> events =
 				this.decoder.decode(Flux.just(stringBuffer(XML)), null, null, Collections.emptyMap());
 
-		TestSubscriber
-				.subscribe(events)
-				.assertNoError()
-				.assertComplete()
-				.assertValuesWith(e -> assertTrue(e.isStartDocument()),
-						e -> assertStartElement(e, "pojo"),
-						e -> assertStartElement(e, "foo"),
-						e -> assertCharacters(e, "foofoo"),
-						e -> assertEndElement(e, "foo"),
-						e -> assertStartElement(e, "bar"),
-						e -> assertCharacters(e, "barbar"),
-						e -> assertEndElement(e, "bar"),
-						e -> assertEndElement(e, "pojo"));
+		ScriptedSubscriber
+				.<XMLEvent>create()
+				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
+				.consumeNextWith(e -> assertStartElement(e, "pojo"))
+				.consumeNextWith(e -> assertStartElement(e, "foo"))
+				.consumeNextWith(e -> assertCharacters(e, "foofoo"))
+				.consumeNextWith(e -> assertEndElement(e, "foo"))
+				.consumeNextWith(e -> assertStartElement(e, "bar"))
+				.consumeNextWith(e -> assertCharacters(e, "barbar"))
+				.consumeNextWith(e -> assertEndElement(e, "bar"))
+				.consumeNextWith(e -> assertEndElement(e, "pojo"))
+				.expectComplete()
+				.verify(events);
 	}
 
 	@Test
@@ -69,19 +69,20 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		Flux<XMLEvent> events =
 				this.decoder.decode(Flux.just(stringBuffer(XML)), null, null, Collections.emptyMap());
 
-		TestSubscriber
-				.subscribe(events)
-				.assertNoError()
-				.assertComplete()
-				.assertValuesWith(e -> assertTrue(e.isStartDocument()),
-						e -> assertStartElement(e, "pojo"),
-						e -> assertStartElement(e, "foo"),
-						e -> assertCharacters(e, "foofoo"),
-						e -> assertEndElement(e, "foo"),
-						e -> assertStartElement(e, "bar"),
-						e -> assertCharacters(e, "barbar"),
-						e -> assertEndElement(e, "bar"), e -> assertEndElement(e, "pojo"),
-						e -> assertTrue(e.isEndDocument()));
+		ScriptedSubscriber
+				.<XMLEvent>create()
+				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
+				.consumeNextWith(e -> assertStartElement(e, "pojo"))
+				.consumeNextWith(e -> assertStartElement(e, "foo"))
+				.consumeNextWith(e -> assertCharacters(e, "foofoo"))
+				.consumeNextWith(e -> assertEndElement(e, "foo"))
+				.consumeNextWith(e -> assertStartElement(e, "bar"))
+				.consumeNextWith(e -> assertCharacters(e, "barbar"))
+				.consumeNextWith(e -> assertEndElement(e, "bar"))
+				.consumeNextWith(e -> assertEndElement(e, "pojo"))
+				.consumeNextWith(e -> assertTrue(e.isEndDocument()))
+				.expectComplete()
+				.verify(events);
 	}
 
 	private static void assertStartElement(XMLEvent event, String expectedLocalName) {

spring-web/src/test/java/org/springframework/http/server/reactive/FlushingIntegrationTests.java

@@ -23,11 +23,11 @@ import org.junit.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.http.codec.BodyExtractors;
-import org.springframework.tests.TestSubscriber;
 import org.springframework.web.client.reactive.ClientRequest;
 import org.springframework.web.client.reactive.WebClient;
 
@@ -56,10 +56,10 @@ public class FlushingIntegrationTests extends AbstractHttpHandlerIntegrationTest
 				.takeUntil(s -> s.endsWith("data1"))
 				.reduce((s1, s2) -> s1 + s2);
 
-		TestSubscriber
-				.subscribe(result)
-				.await(Duration.ofSeconds(5))
-				.assertValues("data0data1");
+		ScriptedSubscriber.<String>create()
+				.expectNext("data0data1")
+				.expectComplete()
+				.verify(result, Duration.ofSeconds(5L));
 	}
 
 	@Override

spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java

@@ -28,6 +28,7 @@ import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
@@ -36,7 +37,6 @@ import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.http.codec.BodyExtractors;
 import org.springframework.http.codec.BodyInserters;
 import org.springframework.http.codec.Pojo;
-import org.springframework.tests.TestSubscriber;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
@@ -70,14 +70,14 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.map(response -> response.headers().asHttpHeaders());
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValuesWith(
+		ScriptedSubscriber.<HttpHeaders>create()
+				.consumeNextWith(
 						httpHeaders -> {
 							assertEquals(MediaType.TEXT_PLAIN, httpHeaders.getContentType());
 							assertEquals(13L, httpHeaders.getContentLength());
 						})
-				.assertComplete();
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -98,10 +98,11 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.then(response -> response.body(toMono(String.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValues("Hello Spring!")
-				.assertComplete();
+		ScriptedSubscriber
+				.<String>create()
+				.expectNext("Hello Spring!")
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -125,10 +126,11 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.then(response -> response.body(toMono(String.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValues(content)
-				.assertComplete();
+		ScriptedSubscriber
+				.<String>create()
+				.expectNext(content)
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -150,10 +152,11 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.then(response -> response.body(toMono(Pojo.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValuesWith(p -> assertEquals("barbar", p.getBar()))
-				.assertComplete();
+		ScriptedSubscriber
+				.<Pojo>create()
+				.consumeNextWith(p -> assertEquals("barbar", p.getBar()))
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -175,13 +178,12 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.flatMap(response -> response.body(toFlux(Pojo.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValuesWith(
-						p -> assertThat(p.getBar(), Matchers.is("bar1")),
-						p -> assertThat(p.getBar(), Matchers.is("bar2")))
-				.assertValueCount(2)
-				.assertComplete();
+		ScriptedSubscriber
+				.<Pojo>create()
+				.consumeNextWith(p -> assertThat(p.getBar(), Matchers.is("bar1")))
+				.consumeNextWith(p -> assertThat(p.getBar(), Matchers.is("bar2")))
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -206,10 +208,11 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.then(response -> response.body(BodyExtractors.toMono(Pojo.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValuesWith(p -> assertEquals("BARBAR", p.getBar()))
-				.assertComplete();
+		ScriptedSubscriber
+				.<Pojo>create()
+				.consumeNextWith(p -> assertEquals("BARBAR", p.getBar()))
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -234,10 +237,11 @@ public class WebClientIntegrationTests {
 				.exchange(request)
 				.then(response -> response.body(toMono(String.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValues("test")
-				.assertComplete();
+		ScriptedSubscriber
+				.<String>create()
+				.expectNext("test")
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -256,12 +260,13 @@ public class WebClientIntegrationTests {
 		Mono<ClientResponse> result = this.webClient
 				.exchange(request);
 
-		TestSubscriber
-				.subscribe(result)
-				.await(Duration.ofSeconds(3))
-				.assertValuesWith(response -> {
+		ScriptedSubscriber
+				.<ClientResponse>create()
+				.consumeNextWith(response -> {
 					assertEquals(HttpStatus.NOT_FOUND, response.statusCode());
-				});
+				})
+				.expectComplete()
+				.verify(result, Duration.ofSeconds(3));
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -287,10 +292,11 @@ public class WebClientIntegrationTests {
 		Mono<String> result = filteredClient.exchange(request)
 				.then(response -> response.body(toMono(String.class)));
 
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValues("Hello Spring!")
-				.assertComplete();
+		ScriptedSubscriber
+				.<String>create()
+				.expectNext("Hello Spring!")
+				.expectComplete()
+				.verify(result);
 
 		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
@@ -302,4 +308,4 @@ public class WebClientIntegrationTests {
 	public void tearDown() throws Exception {
 		this.server.shutdown();
 	}
-}
+}

spring-web/src/test/java/org/springframework/web/server/handler/ResponseStatusExceptionHandlerTests.java

@@ -21,12 +21,12 @@ import java.time.Duration;
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Mono;
+import reactor.test.subscriber.ScriptedSubscriber;
 
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
-import org.springframework.tests.TestSubscriber;
 import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.server.ServerWebExchange;
 import org.springframework.web.server.adapter.DefaultServerWebExchange;
@@ -72,8 +72,10 @@ public class ResponseStatusExceptionHandlerTests {
 		Throwable expected = new IllegalStateException();
 		Mono<Void> mono = this.handler.handle(this.exchange, expected);
 
-		TestSubscriber.subscribe(mono)
-				.assertErrorWith(actual -> assertSame(expected, actual));
+		ScriptedSubscriber
+				.<Void>create()
+				.consumeErrorWith(actual -> assertSame(expected, actual))
+				.verify(mono);
 	}
 
 }