() {
+ @Override
+ public String get() {
+ return errorMessage;
+ }
+ }, conditionSupplier);
+ }
+
+ /**
+ * Create a new {@link TestSubscriber} that requests an unbounded number of elements.
+ * Be sure at least a publisher has subscribed to it via {@link Publisher#subscribe(Subscriber)}
+ * before use assert methods.
+ * @see #subscribe(Publisher)
+ * @param the observed value type
+ * @return a fresh TestSubscriber instance
+ */
+ public static TestSubscriber create() {
+ return new TestSubscriber<>();
+ }
+
+ /**
+ * Create a new {@link TestSubscriber} that requests initially {@code n} elements. You
+ * can then manage the demand with {@link Subscription#request(long)}.
+ * Be sure at least a publisher has subscribed to it via {@link Publisher#subscribe(Subscriber)}
+ * before use assert methods.
+ * @param n Number of elements to request (can be 0 if you want no initial demand).
+ * @see #subscribe(Publisher, long)
+ * @param the observed value type
+ * @return a fresh TestSubscriber instance
+ */
+ public static TestSubscriber create(long n) {
+ return new TestSubscriber<>(n);
+ }
+
+ /**
+ * Create a new {@link TestSubscriber} that requests an unbounded number of elements,
+ * and make the specified {@code publisher} subscribe to it.
+ * @param publisher The publisher to subscribe with
+ * @param the observed value type
+ * @return a fresh TestSubscriber instance
+ */
+ public static TestSubscriber subscribe(Publisher publisher) {
+ TestSubscriber subscriber = new TestSubscriber<>();
+ publisher.subscribe(subscriber);
+ return subscriber;
+ }
+
+ /**
+ * Create a new {@link TestSubscriber} that requests initially {@code n} elements,
+ * and make the specified {@code publisher} subscribe to it. You can then manage the
+ * demand with {@link Subscription#request(long)}.
+ * @param publisher The publisher to subscribe with
+ * @param n Number of elements to request (can be 0 if you want no initial demand).
+ * @param the observed value type
+ * @return a fresh TestSubscriber instance
+ */
+ public static TestSubscriber subscribe(Publisher publisher, long n) {
+ TestSubscriber subscriber = new TestSubscriber<>(n);
+ publisher.subscribe(subscriber);
+ return subscriber;
+ }
+
+// ==============================================================================================================
+// Private constructors
+// ==============================================================================================================
+
+ private TestSubscriber() {
+ this(Long.MAX_VALUE);
+ }
+
+ private TestSubscriber(long n) {
+ if (n < 0) {
+ throw new IllegalArgumentException("initialRequest >= required but it was " + n);
+ }
+ REQUESTED.lazySet(this, n);
+ }
+
+// ==============================================================================================================
+// Configuration
+// ==============================================================================================================
+
+
+ /**
+ * Enable or disabled the values storage. It is enabled by default, and can be disable
+ * in order to be able to perform performance benchmarks or tests with a huge amount
+ * values.
+ * @param enabled enable value storage?
+ * @return this
+ */
+ public final TestSubscriber configureValuesStorage(boolean enabled) {
+ this.valuesStorage = enabled;
+ return this;
+ }
+
+ /**
+ * Configure the timeout in seconds for waiting next values to be received (3 seconds
+ * by default).
+ * @param timeout the new default value timeout duration
+ * @return this
+ */
+ public final TestSubscriber configureValuesTimeout(Duration timeout) {
+ this.valuesTimeout = timeout;
+ return this;
+ }
+
+ /**
+ * Returns the established fusion mode or -1 if it was not enabled
+ *
+ * @return the fusion mode, see Fuseable constants
+ */
+ public final int establishedFusionMode() {
+ return establishedFusionMode;
+ }
+
+// ==============================================================================================================
+// Assertions
+// ==============================================================================================================
+
+ /**
+ * Assert a complete successfully signal has been received.
+ * @return this
+ */
+ public final TestSubscriber assertComplete() {
+ assertNoError();
+ int c = completionCount;
+ if (c == 0) {
+ throw new AssertionError("Not completed", null);
+ }
+ if (c > 1) {
+ throw new AssertionError("Multiple completions: " + c, null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert the specified values have been received. Values storage should be enabled to
+ * use this method.
+ * @param expectedValues the values to assert
+ * @see #configureValuesStorage(boolean)
+ * @return this
+ */
+ public final TestSubscriber assertContainValues(Set expectedValues) {
+ if (!valuesStorage) {
+ throw new IllegalStateException(
+ "Using assertNoValues() requires enabling values storage");
+ }
+ if (expectedValues.size() > values.size()) {
+ throw new AssertionError("Actual contains fewer elements" + values, null);
+ }
+
+ Iterator expected = expectedValues.iterator();
+
+ for (; ; ) {
+ boolean n2 = expected.hasNext();
+ if (n2) {
+ T t2 = expected.next();
+ if (!values.contains(t2)) {
+ throw new AssertionError("The element is not contained in the " +
+ "received resuls" +
+ " = " + valueAndClass(t2), null);
+ }
+ }
+ else{
+ break;
+ }
+ }
+ return this;
+ }
+
+ /**
+ * Assert an error signal has been received.
+ * @return this
+ */
+ public final TestSubscriber assertError() {
+ assertNotComplete();
+ int s = errors.size();
+ if (s == 0) {
+ throw new AssertionError("No error", null);
+ }
+ if (s > 1) {
+ throw new AssertionError("Multiple errors: " + s, null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert an error signal has been received.
+ * @param clazz The class of the exception contained in the error signal
+ * @return this
+ */
+ public final TestSubscriber assertError(Class clazz) {
+ assertNotComplete();
+ int s = errors.size();
+ if (s == 0) {
+ throw new AssertionError("No error", null);
+ }
+ if (s == 1) {
+ Throwable e = errors.get(0);
+ if (!clazz.isInstance(e)) {
+ throw new AssertionError("Error class incompatible: expected = " +
+ clazz + ", actual = " + e, null);
+ }
+ }
+ if (s > 1) {
+ throw new AssertionError("Multiple errors: " + s, null);
+ }
+ return this;
+ }
+
+ public final TestSubscriber assertErrorMessage(String message) {
+ assertNotComplete();
+ int s = errors.size();
+ if (s == 0) {
+ assertionError("No error", null);
+ }
+ if (s == 1) {
+ if (!Objects.equals(message,
+ errors.get(0)
+ .getMessage())) {
+ assertionError("Error class incompatible: expected = \"" + message +
+ "\", actual = \"" + errors.get(0).getMessage() + "\"", null);
+ }
+ }
+ if (s > 1) {
+ assertionError("Multiple errors: " + s, null);
+ }
+
+ return this;
+ }
+
+ /**
+ * Assert an error signal has been received.
+ * @param expectation A method that can verify the exception contained in the error signal
+ * and throw an exception (like an {@link AssertionError}) if the exception is not valid.
+ * @return this
+ */
+ public final TestSubscriber assertErrorWith(Consumer expectation) {
+ assertNotComplete();
+ int s = errors.size();
+ if (s == 0) {
+ throw new AssertionError("No error", null);
+ }
+ if (s == 1) {
+ expectation.accept(errors.get(0));
+ }
+ if (s > 1) {
+ throw new AssertionError("Multiple errors: " + s, null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert that the upstream was a Fuseable source.
+ *
+ * @return this
+ */
+ public final TestSubscriber assertFuseableSource() {
+ if (qs == null) {
+ throw new AssertionError("Upstream was not Fuseable");
+ }
+ return this;
+ }
+
+ /**
+ * Assert that the fusion mode was granted.
+ *
+ * @return this
+ */
+ public final TestSubscriber assertFusionEnabled() {
+ if (establishedFusionMode != Fuseable.SYNC && establishedFusionMode != Fuseable.ASYNC) {
+ throw new AssertionError("Fusion was not enabled");
+ }
+ return this;
+ }
+
+ public final TestSubscriber assertFusionMode(int expectedMode) {
+ if (establishedFusionMode != expectedMode) {
+ throw new AssertionError("Wrong fusion mode: expected: " + fusionModeName(
+ expectedMode) + ", actual: " + fusionModeName(establishedFusionMode));
+ }
+ return this;
+ }
+
+ /**
+ * Assert that the fusion mode was granted.
+ *
+ * @return this
+ */
+ public final TestSubscriber assertFusionRejected() {
+ if (establishedFusionMode != Fuseable.NONE) {
+ throw new AssertionError("Fusion was granted");
+ }
+ return this;
+ }
+
+ /**
+ * Assert no error signal has been received.
+ * @return this
+ */
+ public final TestSubscriber assertNoError() {
+ int s = errors.size();
+ if (s == 1) {
+ Throwable e = errors.get(0);
+ String valueAndClass = e == null ? null : e + " (" + e.getClass().getSimpleName() + ")";
+ throw new AssertionError("Error present: " + valueAndClass, null);
+ }
+ if (s > 1) {
+ throw new AssertionError("Multiple errors: " + s, null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert no values have been received.
+ *
+ * @return this
+ */
+ public final TestSubscriber assertNoValues() {
+ if (valueCount != 0) {
+ throw new AssertionError("No values expected but received: [length = " + values.size() + "] " + values,
+ null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert that the upstream was not a Fuseable source.
+ * @return this
+ */
+ public final TestSubscriber assertNonFuseableSource() {
+ if (qs != null) {
+ throw new AssertionError("Upstream was Fuseable");
+ }
+ return this;
+ }
+
+ /**
+ * Assert no complete successfully signal has been received.
+ * @return this
+ */
+ public final TestSubscriber assertNotComplete() {
+ int c = completionCount;
+ if (c == 1) {
+ throw new AssertionError("Completed", null);
+ }
+ if (c > 1) {
+ throw new AssertionError("Multiple completions: " + c, null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert no subscription occurred.
+ *
+ * @return this
+ */
+ public final TestSubscriber assertNotSubscribed() {
+ int s = subscriptionCount;
+
+ if (s == 1) {
+ throw new AssertionError("OnSubscribe called once", null);
+ }
+ if (s > 1) {
+ throw new AssertionError("OnSubscribe called multiple times: " + s, null);
+ }
+
+ return this;
+ }
+
+ /**
+ * Assert no complete successfully or error signal has been received.
+ * @return this
+ */
+ public final TestSubscriber assertNotTerminated() {
+ if (cdl.getCount() == 0) {
+ throw new AssertionError("Terminated", null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert subscription occurred (once).
+ * @return this
+ */
+ public final TestSubscriber assertSubscribed() {
+ int s = subscriptionCount;
+
+ if (s == 0) {
+ throw new AssertionError("OnSubscribe not called", null);
+ }
+ if (s > 1) {
+ throw new AssertionError("OnSubscribe called multiple times: " + s, null);
+ }
+
+ return this;
+ }
+
+ /**
+ * Assert either complete successfully or error signal has been received.
+ * @return this
+ */
+ public final TestSubscriber assertTerminated() {
+ if (cdl.getCount() != 0) {
+ throw new AssertionError("Not terminated", null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert {@code n} values has been received.
+ *
+ * @param n the expected value count
+ *
+ * @return this
+ */
+ public final TestSubscriber assertValueCount(long n) {
+ if (valueCount != n) {
+ throw new AssertionError("Different value count: expected = " + n + ", actual = " + valueCount,
+ null);
+ }
+ return this;
+ }
+
+ /**
+ * Assert the specified values have been received in the same order read by the
+ * passed {@link Iterable}. Values storage
+ * should be enabled to
+ * use this method.
+ * @param expectedSequence the values to assert
+ * @see #configureValuesStorage(boolean)
+ * @return this
+ */
+ public final TestSubscriber assertValueSequence(Iterable expectedSequence) {
+ if (!valuesStorage) {
+ throw new IllegalStateException("Using assertNoValues() requires enabling values storage");
+ }
+ Iterator actual = values.iterator();
+ Iterator expected = expectedSequence.iterator();
+ int i = 0;
+ for (; ; ) {
+ boolean n1 = actual.hasNext();
+ boolean n2 = expected.hasNext();
+ if (n1 && n2) {
+ T t1 = actual.next();
+ T t2 = expected.next();
+ if (!Objects.equals(t1, t2)) {
+ throw new AssertionError("The element with index " + i + " does not match: expected = " + valueAndClass(t2) + ", actual = "
+ + valueAndClass(
+ t1), null);
+ }
+ i++;
+ } else if (n1 && !n2) {
+ throw new AssertionError("Actual contains more elements" + values, null);
+ } else if (!n1 && n2) {
+ throw new AssertionError("Actual contains fewer elements: " + values, null);
+ } else {
+ break;
+ }
+ }
+ return this;
+ }
+
+ /**
+ * Assert the specified values have been received in the declared order. Values
+ * storage should be enabled to use this method.
+ *
+ * @param expectedValues the values to assert
+ *
+ * @return this
+ *
+ * @see #configureValuesStorage(boolean)
+ */
+ @SafeVarargs
+ public final TestSubscriber assertValues(T... expectedValues) {
+ return assertValueSequence(Arrays.asList(expectedValues));
+ }
+
+ /**
+ * Assert the specified values have been received in the declared order. Values
+ * storage should be enabled to use this method.
+ *
+ * @param expectations One or more methods that can verify the values and throw a
+ * exception (like an {@link AssertionError}) if the value is not valid.
+ *
+ * @return this
+ *
+ * @see #configureValuesStorage(boolean)
+ */
+ @SafeVarargs
+ public final TestSubscriber assertValuesWith(Consumer... expectations) {
+ if (!valuesStorage) {
+ throw new IllegalStateException(
+ "Using assertNoValues() requires enabling values storage");
+ }
+ final int expectedValueCount = expectations.length;
+ if (expectedValueCount != values.size()) {
+ throw new AssertionError("Different value count: expected = " + expectedValueCount + ", actual = " + valueCount, null);
+ }
+ for (int i = 0; i < expectedValueCount; i++) {
+ Consumer consumer = expectations[i];
+ T actualValue = values.get(i);
+ consumer.accept(actualValue);
+ }
+ return this;
+ }
+
+// ==============================================================================================================
+// Await methods
+// ==============================================================================================================
+
+ /**
+ * Blocking method that waits until a complete successfully or error signal is received.
+ * @return this
+ */
+ public final TestSubscriber await() {
+ if (cdl.getCount() == 0) {
+ return this;
+ }
+ try {
+ cdl.await();
+ } catch (InterruptedException ex) {
+ throw new AssertionError("Wait interrupted", ex);
+ }
+ return this;
+ }
+
+ /**
+ * Blocking method that waits until a complete successfully or error signal is received
+ * or until a timeout occurs.
+ * @param timeout The timeout value
+ * @return this
+ */
+ public final TestSubscriber await(Duration timeout) {
+ if (cdl.getCount() == 0) {
+ return this;
+ }
+ try {
+ if (!cdl.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
+ throw new AssertionError("No complete or error signal before timeout");
+ }
+ return this;
+ }
+ catch (InterruptedException ex) {
+ throw new AssertionError("Wait interrupted", ex);
+ }
+ }
+
+ /**
+ * Blocking method that waits until {@code n} next values have been received.
+ *
+ * @param n the value count to assert
+ *
+ * @return this
+ */
+ public final TestSubscriber awaitAndAssertNextValueCount(final long n) {
+ await(valuesTimeout, () -> {
+ if(valuesStorage){
+ return String.format("%d out of %d next values received within %d, " +
+ "values : %s",
+ valueCount - nextValueAssertedCount,
+ n,
+ valuesTimeout.toMillis(),
+ values.toString()
+ );
+ }
+ return String.format("%d out of %d next values received within %d",
+ valueCount - nextValueAssertedCount,
+ n,
+ valuesTimeout.toMillis());
+ }, () -> valueCount >= (nextValueAssertedCount + n));
+ nextValueAssertedCount += n;
+ return this;
+ }
+
+ /**
+ * Blocking method that waits until {@code n} next values have been received (n is the
+ * number of values provided) to assert them.
+ *
+ * @param values the values to assert
+ *
+ * @return this
+ */
+ @SafeVarargs
+ @SuppressWarnings("unchecked")
+ public final TestSubscriber awaitAndAssertNextValues(T... values) {
+ final int expectedNum = values.length;
+ final List> expectations = new ArrayList<>();
+ for (int i = 0; i < expectedNum; i++) {
+ final T expectedValue = values[i];
+ expectations.add(actualValue -> {
+ if (!actualValue.equals(expectedValue)) {
+ throw new AssertionError(String.format(
+ "Expected Next signal: %s, but got: %s",
+ expectedValue,
+ actualValue));
+ }
+ });
+ }
+ awaitAndAssertNextValuesWith(expectations.toArray((Consumer[]) new Consumer[0]));
+ return this;
+ }
+
+ /**
+ * Blocking method that waits until {@code n} next values have been received
+ * (n is the number of expectations provided) to assert them.
+ * @param expectations One or more methods that can verify the values and throw a
+ * exception (like an {@link AssertionError}) if the value is not valid.
+ * @return this
+ */
+ @SafeVarargs
+ public final TestSubscriber awaitAndAssertNextValuesWith(Consumer... expectations) {
+ valuesStorage = true;
+ final int expectedValueCount = expectations.length;
+ await(valuesTimeout, () -> {
+ if(valuesStorage){
+ return String.format("%d out of %d next values received within %d, " +
+ "values : %s",
+ valueCount - nextValueAssertedCount,
+ expectedValueCount,
+ valuesTimeout.toMillis(),
+ values.toString()
+ );
+ }
+ return String.format("%d out of %d next values received within %d ms",
+ valueCount - nextValueAssertedCount,
+ expectedValueCount,
+ valuesTimeout.toMillis());
+ }, () -> valueCount >= (nextValueAssertedCount + expectedValueCount));
+ List nextValuesSnapshot;
+ List empty = new ArrayList<>();
+ for(;;){
+ nextValuesSnapshot = values;
+ if(NEXT_VALUES.compareAndSet(this, values, empty)){
+ break;
+ }
+ }
+ if (nextValuesSnapshot.size() < expectedValueCount) {
+ throw new AssertionError(String.format("Expected %d number of signals but received %d",
+ expectedValueCount,
+ nextValuesSnapshot.size()));
+ }
+ for (int i = 0; i < expectedValueCount; i++) {
+ Consumer consumer = expectations[i];
+ T actualValue = nextValuesSnapshot.get(i);
+ consumer.accept(actualValue);
+ }
+ nextValueAssertedCount += expectedValueCount;
+ return this;
+ }
+
+// ==============================================================================================================
+// Overrides
+// ==============================================================================================================
+
+ @Override
+ public void cancel() {
+ Subscription a = s;
+ if (a != Operators.cancelledSubscription()) {
+ a = S.getAndSet(this, Operators.cancelledSubscription());
+ if (a != null && a != Operators.cancelledSubscription()) {
+ a.cancel();
+ }
+ }
+ }
+
+ @Override
+ public final boolean isCancelled() {
+ return s == Operators.cancelledSubscription();
+ }
+
+ @Override
+ public final boolean isStarted() {
+ return s != null;
+ }
+
+ @Override
+ public final boolean isTerminated() {
+ return isCancelled();
+ }
+
+ @Override
+ public void onComplete() {
+ completionCount++;
+ cdl.countDown();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errors.add(t);
+ cdl.countDown();
+ }
+
+ @Override
+ public void onNext(T t) {
+ if (establishedFusionMode == Fuseable.ASYNC) {
+ for (; ; ) {
+ t = qs.poll();
+ if (t == null) {
+ break;
+ }
+ valueCount++;
+ if (valuesStorage) {
+ List nextValuesSnapshot;
+ for (; ; ) {
+ nextValuesSnapshot = values;
+ nextValuesSnapshot.add(t);
+ if (NEXT_VALUES.compareAndSet(this,
+ nextValuesSnapshot,
+ nextValuesSnapshot)) {
+ break;
+ }
+ }
+ }
+ }
+ }
+ else {
+ valueCount++;
+ if (valuesStorage) {
+ List nextValuesSnapshot;
+ for (; ; ) {
+ nextValuesSnapshot = values;
+ nextValuesSnapshot.add(t);
+ if (NEXT_VALUES.compareAndSet(this,
+ nextValuesSnapshot,
+ nextValuesSnapshot)) {
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void onSubscribe(Subscription s) {
+ subscriptionCount++;
+ int requestMode = requestedFusionMode;
+ if (requestMode >= 0) {
+ if (!setWithoutRequesting(s)) {
+ if (!isCancelled()) {
+ errors.add(new IllegalStateException("Subscription already set: " +
+ subscriptionCount));
+ }
+ } else {
+ if (s instanceof Fuseable.QueueSubscription) {
+ this.qs = (Fuseable.QueueSubscription)s;
+
+ int m = qs.requestFusion(requestMode);
+ establishedFusionMode = m;
+
+ if (m == Fuseable.SYNC) {
+ for (;;) {
+ T v = qs.poll();
+ if (v == null) {
+ onComplete();
+ break;
+ }
+
+ onNext(v);
+ }
+ }
+ else {
+ requestDeferred();
+ }
+ }
+ else {
+ requestDeferred();
+ }
+ }
+ } else {
+ if (!set(s)) {
+ if (!isCancelled()) {
+ errors.add(new IllegalStateException("Subscription already set: " +
+ subscriptionCount));
+ }
+ }
+ }
+ }
+
+ @Override
+ public void request(long n) {
+ if (Operators.validate(n)) {
+ if (establishedFusionMode != Fuseable.SYNC) {
+ normalRequest(n);
+ }
+ }
+ }
+
+ @Override
+ public final long requestedFromDownstream() {
+ return requested;
+ }
+
+ /**
+ * Setup what fusion mode should be requested from the incomining
+ * Subscription if it happens to be QueueSubscription
+ * @param requestMode the mode to request, see Fuseable constants
+ * @return this
+ */
+ public final TestSubscriber requestedFusionMode(int requestMode) {
+ this.requestedFusionMode = requestMode;
+ return this;
+ }
+
+ @Override
+ public Subscription upstream() {
+ return s;
+ }
+
+
+// ==============================================================================================================
+// Non public methods
+// ==============================================================================================================
+
+ protected final void normalRequest(long n) {
+ Subscription a = s;
+ if (a != null) {
+ a.request(n);
+ } else {
+ Operators.addAndGet(REQUESTED, this, n);
+
+ a = s;
+
+ if (a != null) {
+ long r = REQUESTED.getAndSet(this, 0L);
+
+ if (r != 0L) {
+ a.request(r);
+ }
+ }
+ }
+ }
+
+ /**
+ * Requests the deferred amount if not zero.
+ */
+ protected final void requestDeferred() {
+ long r = REQUESTED.getAndSet(this, 0L);
+
+ if (r != 0L) {
+ s.request(r);
+ }
+ }
+
+ /**
+ * Atomically sets the single subscription and requests the missed amount from it.
+ *
+ * @param s
+ * @return false if this arbiter is cancelled or there was a subscription already set
+ */
+ protected final boolean set(Subscription s) {
+ Objects.requireNonNull(s, "s");
+ Subscription a = this.s;
+ if (a == Operators.cancelledSubscription()) {
+ s.cancel();
+ return false;
+ }
+ if (a != null) {
+ s.cancel();
+ Operators.reportSubscriptionSet();
+ return false;
+ }
+
+ if (S.compareAndSet(this, null, s)) {
+
+ long r = REQUESTED.getAndSet(this, 0L);
+
+ if (r != 0L) {
+ s.request(r);
+ }
+
+ return true;
+ }
+
+ a = this.s;
+
+ if (a != Operators.cancelledSubscription()) {
+ s.cancel();
+ return false;
+ }
+
+ Operators.reportSubscriptionSet();
+ return false;
+ }
+
+ /**
+ * Sets the Subscription once but does not request anything.
+ * @param s the Subscription to set
+ * @return true if successful, false if the current subscription is not null
+ */
+ protected final boolean setWithoutRequesting(Subscription s) {
+ Objects.requireNonNull(s, "s");
+ for (;;) {
+ Subscription a = this.s;
+ if (a == Operators.cancelledSubscription()) {
+ s.cancel();
+ return false;
+ }
+ if (a != null) {
+ s.cancel();
+ Operators.reportSubscriptionSet();
+ return false;
+ }
+
+ if (S.compareAndSet(this, null, s)) {
+ return true;
+ }
+ }
+ }
+
+ /**
+ * Prepares and throws an AssertionError exception based on the message, cause, the
+ * active state and the potential errors so far.
+ *
+ * @param message the message
+ * @param cause the optional Throwable cause
+ *
+ * @throws AssertionError as expected
+ */
+ protected final void assertionError(String message, Throwable cause) {
+ StringBuilder b = new StringBuilder();
+
+ if (cdl.getCount() != 0) {
+ b.append("(active) ");
+ }
+ b.append(message);
+
+ List err = errors;
+ if (!err.isEmpty()) {
+ b.append(" (+ ")
+ .append(err.size())
+ .append(" errors)");
+ }
+ AssertionError e = new AssertionError(b.toString(), cause);
+
+ for (Throwable t : err) {
+ e.addSuppressed(t);
+ }
+
+ throw e;
+ }
+
+ protected final String fusionModeName(int mode) {
+ switch (mode) {
+ case -1:
+ return "Disabled";
+ case Fuseable.NONE:
+ return "None";
+ case Fuseable.SYNC:
+ return "Sync";
+ case Fuseable.ASYNC:
+ return "Async";
+ default:
+ return "Unknown(" + mode + ")";
+ }
+ }
+
+ protected final String valueAndClass(Object o) {
+ if (o == null) {
+ return null;
+ }
+ return o + " (" + o.getClass().getSimpleName() + ")";
+ }
+
+}
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java
index 414c5023b2..7b1cb9e2e6 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java
@@ -24,7 +24,6 @@ import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
@@ -39,6 +38,7 @@ import org.springframework.http.codec.EncoderHttpMessageWriter;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.stereotype.Controller;
+import org.springframework.tests.TestSubscriber;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/ResponseStatusExceptionHandlerTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/ResponseStatusExceptionHandlerTests.java
index f8ccd90416..6afddbaba1 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/ResponseStatusExceptionHandlerTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/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.TestSubscriber;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.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;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/HandlerMethodMappingTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/HandlerMethodMappingTests.java
index 6259b814ec..b14ba1aab1 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/HandlerMethodMappingTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/HandlerMethodMappingTests.java
@@ -27,13 +27,13 @@ import java.util.Set;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Controller;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.PathMatcher;
import org.springframework.web.bind.annotation.RequestMapping;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/InvocableHandlerMethodTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/InvocableHandlerMethodTests.java
index 28fddee9d2..646852bd6e 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/InvocableHandlerMethodTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/InvocableHandlerMethodTests.java
@@ -22,11 +22,11 @@ import java.util.Optional;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
+import org.springframework.tests.TestSubscriber;
import org.springframework.ui.ExtendedModelMap;
import org.springframework.ui.ModelMap;
import org.springframework.web.reactive.HandlerResult;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/RequestMappingInfoHandlerMappingTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/RequestMappingInfoHandlerMappingTests.java
index 7d843bfb3c..982a704947 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/RequestMappingInfoHandlerMappingTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/RequestMappingInfoHandlerMappingTests.java
@@ -30,7 +30,6 @@ import java.util.function.Consumer;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.AnnotationUtils;
@@ -41,6 +40,7 @@ import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Controller;
+import org.springframework.tests.TestSubscriber;
import org.springframework.ui.ExtendedModelMap;
import org.springframework.ui.ModelMap;
import org.springframework.util.MultiValueMap;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/CookieValueMethodArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/CookieValueMethodArgumentResolverTests.java
index 204f779b34..2f5c27f206 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/CookieValueMethodArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/CookieValueMethodArgumentResolverTests.java
@@ -22,7 +22,6 @@ import java.net.URI;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.MethodParameter;
@@ -34,6 +33,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.web.bind.annotation.CookieValue;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.ServerWebInputException;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/HttpEntityArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/HttpEntityArgumentResolverTests.java
index d6d689daa0..027d46c903 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/HttpEntityArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/HttpEntityArgumentResolverTests.java
@@ -28,7 +28,6 @@ import org.junit.Test;
import reactor.adapter.RxJava1Adapter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import rx.Observable;
import rx.Single;
@@ -46,6 +45,7 @@ import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.ui.ExtendedModelMap;
+import org.springframework.tests.TestSubscriber;
import org.springframework.validation.Validator;
import org.springframework.web.reactive.result.ResolvableMethod;
import org.springframework.web.server.ServerWebExchange;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageReaderArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageReaderArgumentResolverTests.java
index 9ba974f99a..c34ea1f47c 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageReaderArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageReaderArgumentResolverTests.java
@@ -35,7 +35,6 @@ import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import rx.Observable;
import rx.Single;
@@ -51,6 +50,7 @@ import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
+import org.springframework.tests.TestSubscriber;
import org.springframework.validation.Errors;
import org.springframework.validation.Validator;
import org.springframework.validation.annotation.Validated;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageWriterResultHandlerTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageWriterResultHandlerTests.java
index 5ea29597f6..f39d68fed2 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageWriterResultHandlerTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageWriterResultHandlerTests.java
@@ -33,7 +33,6 @@ import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import rx.Completable;
import rx.Observable;
@@ -53,6 +52,7 @@ import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
import org.springframework.web.reactive.accept.RequestedContentTypeResolverBuilder;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/PathVariableMethodArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/PathVariableMethodArgumentResolverTests.java
index c4007f55da..afb9290a7e 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/PathVariableMethodArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/PathVariableMethodArgumentResolverTests.java
@@ -24,7 +24,6 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.core.MethodParameter;
import org.springframework.core.convert.ConversionService;
@@ -33,6 +32,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.reactive.HandlerMapping;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestAttributeMethodArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestAttributeMethodArgumentResolverTests.java
index c8cf6ba285..c6d7902c2d 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestAttributeMethodArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestAttributeMethodArgumentResolverTests.java
@@ -22,7 +22,6 @@ import java.util.Optional;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.DefaultParameterNameDiscoverer;
@@ -35,6 +34,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.ReflectionUtils;
import org.springframework.web.bind.annotation.RequestAttribute;
import org.springframework.web.server.ServerWebExchange;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolverTests.java
index 9d64e08413..e8d9e896de 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolverTests.java
@@ -30,7 +30,6 @@ import org.junit.Test;
import reactor.adapter.RxJava1Adapter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import rx.Observable;
import rx.Single;
@@ -44,6 +43,7 @@ import org.springframework.http.codec.DecoderHttpMessageReader;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
+import org.springframework.tests.TestSubscriber;
import org.springframework.ui.ExtendedModelMap;
import org.springframework.validation.Validator;
import org.springframework.web.bind.annotation.RequestBody;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestHeaderMethodArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestHeaderMethodArgumentResolverTests.java
index 1f795dc34a..d22a2179d2 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestHeaderMethodArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestHeaderMethodArgumentResolverTests.java
@@ -27,7 +27,6 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.MethodParameter;
@@ -38,6 +37,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.ReflectionUtils;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.server.ServerWebExchange;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestParamMethodArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestParamMethodArgumentResolverTests.java
index 3f92c73bc4..c9226a1731 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestParamMethodArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestParamMethodArgumentResolverTests.java
@@ -25,7 +25,6 @@ import java.util.Optional;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
import org.springframework.core.MethodParameter;
@@ -38,6 +37,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.ReflectionUtils;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.server.ServerWebExchange;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseEntityResultHandlerTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseEntityResultHandlerTests.java
index 6278c150bf..40fe1fb54d 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseEntityResultHandlerTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseEntityResultHandlerTests.java
@@ -27,7 +27,6 @@ import java.util.concurrent.CompletableFuture;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import rx.Completable;
import rx.Single;
@@ -47,6 +46,7 @@ import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SessionAttributeMethodArgumentResolverTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SessionAttributeMethodArgumentResolverTests.java
index a7af773e13..6fe286945d 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SessionAttributeMethodArgumentResolverTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SessionAttributeMethodArgumentResolverTests.java
@@ -22,7 +22,6 @@ import java.util.Optional;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.DefaultParameterNameDiscoverer;
@@ -35,6 +34,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.tests.TestSubscriber;
import org.springframework.util.ReflectionUtils;
import org.springframework.web.bind.annotation.SessionAttribute;
import org.springframework.web.server.ServerWebExchange;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java
index dc7c38393b..dbf259b82f 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java
@@ -22,7 +22,6 @@ import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
@@ -32,6 +31,7 @@ import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.codec.SseEvent;
import org.springframework.http.server.reactive.AbstractHttpHandlerIntegrationTests;
import org.springframework.http.server.reactive.HttpHandler;
+import org.springframework.tests.TestSubscriber;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.reactive.WebClient;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/HttpMessageWriterViewTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/HttpMessageWriterViewTests.java
index ff26409785..fbf4c1ea53 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/HttpMessageWriterViewTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/HttpMessageWriterViewTests.java
@@ -28,7 +28,6 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
-import reactor.test.TestSubscriber;
import org.springframework.core.MethodParameter;
import org.springframework.core.codec.CharSequenceEncoder;
@@ -39,6 +38,7 @@ import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
+import org.springframework.tests.TestSubscriber;
import org.springframework.ui.ExtendedModelMap;
import org.springframework.ui.ModelMap;
import org.springframework.util.MimeType;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/ViewResolutionResultHandlerTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/ViewResolutionResultHandlerTests.java
index ff31218d77..8bc1fe11a5 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/ViewResolutionResultHandlerTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/ViewResolutionResultHandlerTests.java
@@ -32,7 +32,6 @@ import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import rx.Completable;
import rx.Single;
@@ -47,6 +46,7 @@ import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.tests.TestSubscriber;
import org.springframework.ui.ExtendedModelMap;
import org.springframework.ui.Model;
import org.springframework.ui.ModelMap;
diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerViewTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerViewTests.java
index e37765cff3..6181eb4ba8 100644
--- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerViewTests.java
+++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerViewTests.java
@@ -25,7 +25,6 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
-import reactor.test.TestSubscriber;
import org.springframework.context.ApplicationContextException;
import org.springframework.context.support.GenericApplicationContext;
@@ -36,6 +35,7 @@ import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.ui.ExtendedModelMap;
import org.springframework.ui.ModelMap;
+import org.springframework.tests.TestSubscriber;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.adapter.DefaultServerWebExchange;
diff --git a/spring-web/src/test/java/org/springframework/http/codec/SseEventHttpMessageWriterTests.java b/spring-web/src/test/java/org/springframework/http/codec/SseEventHttpMessageWriterTests.java
index 312cc1be01..24d8bf5068 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/SseEventHttpMessageWriterTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/SseEventHttpMessageWriterTests.java
@@ -22,7 +22,6 @@ import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -30,6 +29,7 @@ import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.http.server.reactive.MockServerHttpResponse;
+import org.springframework.tests.TestSubscriber;
import static org.junit.Assert.*;
diff --git a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java
index 63b0beae05..b68c512ef0 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java
@@ -24,13 +24,13 @@ import com.fasterxml.jackson.annotation.JsonView;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
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.*;
diff --git a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
index a071cc8cdb..1ff19cbc6e 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
@@ -22,13 +22,13 @@ import com.fasterxml.jackson.annotation.JsonView;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
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.*;
diff --git a/spring-web/src/test/java/org/springframework/http/codec/json/JsonObjectDecoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/json/JsonObjectDecoderTests.java
index c8953bc0f0..2176b94d1c 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/json/JsonObjectDecoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/JsonObjectDecoderTests.java
@@ -20,10 +20,10 @@ import java.nio.charset.StandardCharsets;
import org.junit.Test;
import reactor.core.publisher.Flux;
-import reactor.test.TestSubscriber;
import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.tests.TestSubscriber;
/**
* @author Sebastien Deleuze
diff --git a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlDecoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlDecoderTests.java
index be04ffab0f..3a773b11e8 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlDecoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlDecoderTests.java
@@ -22,7 +22,6 @@ import javax.xml.stream.events.XMLEvent;
import org.junit.Test;
import reactor.core.publisher.Flux;
-import reactor.test.TestSubscriber;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -35,6 +34,7 @@ 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.*;
diff --git a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
index c05ed2c6cc..939fb54284 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
@@ -20,7 +20,6 @@ import java.nio.charset.StandardCharsets;
import org.junit.Test;
import reactor.core.publisher.Flux;
-import reactor.test.TestSubscriber;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
@@ -29,6 +28,7 @@ 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.*;
diff --git a/spring-web/src/test/java/org/springframework/http/codec/xml/XmlEventDecoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/xml/XmlEventDecoderTests.java
index b45b85f943..a84de35bff 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/xml/XmlEventDecoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/xml/XmlEventDecoderTests.java
@@ -20,9 +20,9 @@ import javax.xml.stream.events.XMLEvent;
import org.junit.Test;
import reactor.core.publisher.Flux;
-import reactor.test.TestSubscriber;
import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.tests.TestSubscriber;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
diff --git a/spring-web/src/test/java/org/springframework/http/server/reactive/FlushingIntegrationTests.java b/spring-web/src/test/java/org/springframework/http/server/reactive/FlushingIntegrationTests.java
index ac268fadb3..677f2252fc 100644
--- a/spring-web/src/test/java/org/springframework/http/server/reactive/FlushingIntegrationTests.java
+++ b/spring-web/src/test/java/org/springframework/http/server/reactive/FlushingIntegrationTests.java
@@ -21,10 +21,10 @@ import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.tests.TestSubscriber;
import org.springframework.web.client.reactive.ClientWebRequestBuilders;
import org.springframework.web.client.reactive.ResponseExtractors;
import org.springframework.web.client.reactive.WebClient;
diff --git a/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultResponseErrorHandlerTests.java b/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultResponseErrorHandlerTests.java
index f142f1ff1d..d6a30d423d 100644
--- a/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultResponseErrorHandlerTests.java
+++ b/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultResponseErrorHandlerTests.java
@@ -6,7 +6,6 @@ import java.util.List;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
-import reactor.test.TestSubscriber;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.DataBuffer;
@@ -17,6 +16,7 @@ import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.codec.DecoderHttpMessageReader;
import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.tests.TestSubscriber;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
diff --git a/spring-web/src/test/java/org/springframework/web/client/reactive/ResponseExtractorsTests.java b/spring-web/src/test/java/org/springframework/web/client/reactive/ResponseExtractorsTests.java
index 223724a02a..a54adfb545 100644
--- a/spring-web/src/test/java/org/springframework/web/client/reactive/ResponseExtractorsTests.java
+++ b/spring-web/src/test/java/org/springframework/web/client/reactive/ResponseExtractorsTests.java
@@ -25,7 +25,6 @@ import org.junit.Test;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.DataBuffer;
@@ -38,6 +37,7 @@ import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.codec.DecoderHttpMessageReader;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
+import org.springframework.tests.TestSubscriber;
import static org.hamcrest.CoreMatchers.*;
import static org.junit.Assert.*;
diff --git a/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java b/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java
index f6c095124e..167519616d 100644
--- a/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java
+++ b/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java
@@ -33,13 +33,13 @@ import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.codec.Pojo;
+import org.springframework.tests.TestSubscriber;
/**
* {@link WebClient} integration tests with the {@code Flux} and {@code Mono} API.