1.x: update AssertableSubscriber API and add Javadoc (#4824)

Author: akarnokd

src/main/java/rx/Completable.java

@@ -2365,7 +2365,7 @@ public void call() {
             }
         });
     }
-    
+
     // -------------------------------------------------------------------------
     // Fluent test support, super handy and reduces test preparation boilerplate
     // -------------------------------------------------------------------------

src/main/java/rx/Observable.java

@@ -12642,7 +12642,7 @@ public final <T2, R> Observable<R> zipWith(Iterable<? extends T2> other, Func2<?
     public final <T2, R> Observable<R> zipWith(Observable<? extends T2> other, Func2<? super T, ? super T2, ? extends R> zipFunction) {
         return (Observable<R>)zip(this, other, zipFunction);
     }
-    
+
     // -------------------------------------------------------------------------
     // Fluent test support, super handy and reduces test preparation boilerplate
     // -------------------------------------------------------------------------
@@ -12664,7 +12664,7 @@ public final AssertableSubscriber<T> test() {
         subscribe(ts);
         return ts;
     }
-    
+
     /**
      * Creates an AssertableSubscriber with the initial request amount and subscribes
      * it to this Observable.
@@ -12675,6 +12675,7 @@ public final AssertableSubscriber<T> test() {
      *  <dd>{@code test} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
      * @return the new AssertableSubscriber instance
+     * @param initialRequestAmount the amount to request from upstream upfront, non-negative (not verified)
      * @since 1.2.3
      */
     @Experimental

src/main/java/rx/internal/observers/AssertableSubscriberObservable.java

@@ -135,8 +135,11 @@ public AssertableSubscriber<T> assertReceivedOnNext(List<T> items) {
      * @see rx.observers.AssertableSubscriber#awaitValueCount(int, long, java.util.concurrent.TimeUnit)
      */
     @Override
-    public final boolean awaitValueCount(int expected, long timeout, TimeUnit unit) {
-        return ts.awaitValueCount(expected, timeout, unit);
+    public final AssertableSubscriber<T> awaitValueCount(int expected, long timeout, TimeUnit unit) {
+        if (!ts.awaitValueCount(expected, timeout, unit)) {
+            throw new AssertionError("Did not receive enough values in time. Expected: " + expected + ", Actual: " + ts.getValueCount());
+        }
+        return this;
     }
 
     /* (non-Javadoc)
@@ -282,7 +285,7 @@ public AssertableSubscriber<T> assertValue(T value) {
         ts.assertValue(value);
         return this;
     }
-    
+
     /* (non-Javadoc)
      * @see rx.observers.AssertableSubscriber#assertValuesAndClear(T, T)
      */
@@ -307,4 +310,34 @@ public String toString() {
         return ts.toString();
     }
 
+    @Override
+    public final AssertableSubscriber<T> assertResult(T... values) {
+        ts.assertValues(values);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+        return this;
+    }
+
+    @Override
+    public final AssertableSubscriber<T> assertFailure(Class<? extends Throwable> errorClass, T... values) {
+        ts.assertValues(values);
+        ts.assertError(errorClass);
+        ts.assertNotCompleted();
+        return this;
+    }
+
+    @Override
+    public final AssertableSubscriber<T> assertFailureAndMessage(Class<? extends Throwable> errorClass, String message,
+            T... values) {
+        ts.assertValues(values);
+        ts.assertError(errorClass);
+        ts.assertNotCompleted();
+
+        String actualMessage = ts.getOnErrorEvents().get(0).getMessage();
+        if (!(actualMessage == message || (message != null && message.equals(actualMessage)))) {
+            throw new AssertionError("Error message differs. Expected: \'" + message + "\', Received: \'" + actualMessage + "\'");
+        }
+
+        return this;
+    }
 }

src/main/java/rx/observers/AssertableSubscriber.java

@@ -18,70 +18,260 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
-import rx.Observer;
-import rx.Producer;
+import rx.*;
+import rx.annotations.Experimental;
 import rx.functions.Action0;
 
-public interface AssertableSubscriber<T> extends Observer<T> {
+/**
+ * Interface for asserting the state of a sequence under testing with a {@code test()}
+ * method of a reactive base class.
+ * <p>
+ * This interface is not intended to be implemented outside of RxJava.
+ * <p>
+ * This interface extends {@link Observer} and allows injecting onXXX signals into
+ * the testing process.
+ * @param <T> the value type consumed by this Observer
+ * @since 1.2.3
+ */
+@Experimental
+public interface AssertableSubscriber<T> extends Observer<T>, Subscription {
 
+    /**
+     * Allows manually calling the {@code onStart} method of the underlying Subscriber.
+     */
     void onStart();
 
+    /**
+     * Allows manually calling the {@code setProducer} method of the underlying Subscriber.
+     * @param p the producer to use, not null
+     */
     void setProducer(Producer p);
 
+    /**
+     * Returns the number of {@code onCompleted} signals received by this Observer.
+     * @return the number of {@code onCompleted} signals received
+     */
     int getCompletions();
 
+    /**
+     * Returns a list of received {@code onError} signals.
+     * @return this
+     */
     List<Throwable> getOnErrorEvents();
 
+    /**
+     * Returns the number of {@code onNext} signals received by this Observer in
+     * a thread-safe manner; one can read up to this number of elements from
+     * the {@code List} returned by {@link #getOnNextEvents()}.
+     * @return the number of {@code onNext} signals received.
+     */
     int getValueCount();
 
+    /**
+     * Requests the specified amount of items from upstream.
+     * @param n the amount requested, non-negative
+     * @return this
+     */
     AssertableSubscriber<T> requestMore(long n);
 
+    /**
+     * Returns the list of received {@code onNext} events.
+     * <p>If the sequence hasn't completed yet and is asynchronous, use the
+     * {@link #getValueCount()} method to determine how many elements are safe
+     * to be read from the list returned by this method.
+     * @return the List of received {@code onNext} events.
+     */
     List<T> getOnNextEvents();
 
+    /**
+     * Assert that this Observer received the given list of items as {@code onNext} signals
+     * in the same order and with the default null-safe object equals comparison.
+     * @param items the List of items expected
+     * @return this
+     */
     AssertableSubscriber<T> assertReceivedOnNext(List<T> items);
 
-    boolean awaitValueCount(int expected, long timeout, TimeUnit unit);
+    /**
+     * Assert that this Observer receives at least the given number of {@code onNext}
+     * signals within the specified timeout period.
+     * <p>
+     * Note that it is possible the AssertionError thrown by this method will
+     * contain an actual value &gt;= to the expected one in case there is an emission
+     * race or unexpected delay on the emitter side. In this case, increase the timeout
+     * amount to avoid false positives.
+     *
+     * @param expected the expected (at least) number of {@code onNext} signals
+     * @param timeout the timeout to wait to receive the given number of {@code onNext} events
+     * @param unit the time unit
+     * @return this
+     */
+    AssertableSubscriber<T> awaitValueCount(int expected, long timeout, TimeUnit unit);
 
+    /**
+     * Assert that this Observer received either an {@code onError} or {@code onCompleted} signal.
+     * @return this
+     */
     AssertableSubscriber<T> assertTerminalEvent();
 
+    /**
+     * Assert that this Observer has been unsubscribed via {@code unsubscribe()} or by a wrapping
+     * {@code SafeSubscriber}.
+     * @return this
+     */
     AssertableSubscriber<T> assertUnsubscribed();
 
+    /**
+     * Assert that this Observer has not received any {@code onError} signal.
+     * @return this
+     */
     AssertableSubscriber<T> assertNoErrors();
 
+    /**
+     * Waits for an {@code onError} or {code onCompleted} terminal event indefinitely.
+     * @return this
+     */
     AssertableSubscriber<T> awaitTerminalEvent();
 
+
+    /**
+     * Waits for an {@code onError} or {code onCompleted} terminal event for the given
+     * amount of timeout.
+     * @param timeout the time to wait for the terminal event
+     * @param unit the time unit of the wait time
+     * @return this
+     */
     AssertableSubscriber<T> awaitTerminalEvent(long timeout, TimeUnit unit);
 
+    /**
+     * Waits for an {@code onError} or {code onCompleted} terminal event for the given
+     * amount of timeout and unsubscribes the sequence if the timeout passed or the
+     * wait itself is interrupted.
+     * @param timeout the time to wait for the terminal event
+     * @param unit the time unit of the wait time
+     * @return this
+     */
     AssertableSubscriber<T> awaitTerminalEventAndUnsubscribeOnTimeout(long timeout,
             TimeUnit unit);
 
+    /**
+     * Returns the Thread that has called the last {@code onNext}, {@code onError} or
+     * {@code onCompleted} methods of this Observer.
+     * @return this
+     */
     Thread getLastSeenThread();
 
+    /**
+     * Assert that this Observer received exaclty one {@code onCompleted} signal.
+     * @return this
+     */
     AssertableSubscriber<T> assertCompleted();
 
+    /**
+     * Assert that this Observer received no {@code onCompleted} signal.
+     * @return this
+     */
     AssertableSubscriber<T> assertNotCompleted();
 
+    /**
+     * Assert that this Observer received one {@code onError} signal with
+     * the given subclass of a Throwable as type.
+     * @param clazz the expected type of the {@code onError} signal received
+     * @return this
+     */
     AssertableSubscriber<T> assertError(Class<? extends Throwable> clazz);
 
+    /**
+     * Assert that this Observer received one {@code onError} signal with the
+     * object-equals of the given Throwable instance
+     * @param throwable the Throwable instance expected
+     * @return this
+     */
     AssertableSubscriber<T> assertError(Throwable throwable);
 
+    /**
+     * Assert that no {@code onError} or {@code onCompleted} signals were received so far.
+     * @return this
+     */
     AssertableSubscriber<T> assertNoTerminalEvent();
 
+    /**
+     * Assert that no {@code onNext} signals were received so far.
+     * @return this
+     */
     AssertableSubscriber<T> assertNoValues();
 
+    /**
+     * Assert that this Observer received exactly the given count of
+     * {@code onNext} signals.
+     * @param count the expected number of {@code onNext} signals
+     * @return this
+     */
     AssertableSubscriber<T> assertValueCount(int count);
 
+    /**
+     * Assert that this Observer received exactly the given expected values
+     * (compared via null-safe object equals) in the given order.
+     * @param values the expected values
+     * @return this
+     */
     AssertableSubscriber<T> assertValues(T... values);
 
+    /**
+     * Assert that this Observer received exactly the given single expected value
+     * (compared via null-safe object equals).
+     * @param value the single value expected
+     * @return this
+     */
     AssertableSubscriber<T> assertValue(T value);
 
+    /**
+     * Assert that this Observer received exactly the given values (compared via
+     * null-safe object equals) and if so, clears the internal buffer of the
+     * underlying Subscriber of these values.
+     * @param expectedFirstValue the first value expected
+     * @param expectedRestValues the rest of the values expected
+     * @return this
+     */
     AssertableSubscriber<T> assertValuesAndClear(T expectedFirstValue,
             T... expectedRestValues);
 
+    /**
+     * Performs an action given by the Action0 callback in a fluent manner.
+     * @param action the action to perform, not null
+     * @return this
+     */
     AssertableSubscriber<T> perform(Action0 action);
-    
+
+    @Override
     void unsubscribe();
-    
+
+    @Override
     boolean isUnsubscribed();
 
+    /**
+     * Assert that this Observer received the specified items in the given order followed
+     * by a completion signal and no errors.
+     * @param values the values expected
+     * @return this
+     */
+    AssertableSubscriber<T> assertResult(T... values);
+
+    /**
+     * Assert that this Observer received the specified items in the given order followed
+     * by an error signal of the given type (but no completion signal).
+     * @param errorClass the expected Throwable subclass type
+     * @param values the expected values
+     * @return this
+     */
+    AssertableSubscriber<T> assertFailure(Class<? extends Throwable> errorClass, T... values);
+
+    /**
+     * Assert that this Observer received the specified items in the given order followed
+     * by an error signal of the given type and with the exact error message (but no completion signal).
+     * @param errorClass the expected Throwable subclass type
+     * @param message the expected error message returned by {@link Throwable#getMessage()}
+     * @param values the expected values
+     * @return this
+     */
+    AssertableSubscriber<T> assertFailureAndMessage(Class<? extends Throwable> errorClass, String message, T... values);
 }