() {
+
+ @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
+ @SuppressWarnings("varargs")
+ 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", "rawtypes" })
+ 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() + ")";
+ }
+
+}