TestObserver

rxjava-core/src/main/java/rx/Notification.java

@@ -26,12 +26,35 @@ public class Notification<T> {
     private final Throwable throwable;
     private final T value;
 
+    public static <T> Notification<T> createOnNext(T t) {
+        return new Notification<T>(Kind.OnNext, t, null);
+    }
+
+    public static <T> Notification<T> createOnError(Throwable e) {
+        return new Notification<T>(Kind.OnError, null, e);
+    }
+
+    public static <T> Notification<T> createOnCompleted() {
+        return new Notification<T>(Kind.OnCompleted, null, null);
+    }
+
+    public static <T> Notification<T> createOnCompleted(Class<T> type) {
+        return new Notification<T>(Kind.OnCompleted, null, null);
+    }
+
+    private Notification(Kind kind, T value, Throwable e) {
+        this.value = value;
+        this.throwable = e;
+        this.kind = kind;
+    }
+
     /**
      * A constructor used to represent an onNext notification.
      * 
      * @param value
      *            The data passed to the onNext method.
      */
+    @Deprecated
     public Notification(T value) {
         this.value = value;
         this.throwable = null;
@@ -43,7 +66,9 @@ public Notification(T value) {
      * 
      * @param exception
      *            The exception passed to the onError notification.
+     * @deprecated Because type Throwable can't disambiguate the constructors if both onNext and onError are type "Throwable"
      */
+    @Deprecated
     public Notification(Throwable exception) {
         this.throwable = exception;
         this.value = null;
@@ -53,6 +78,7 @@ public Notification(Throwable exception) {
     /**
      * A constructor used to represent an onCompleted notification.
      */
+    @Deprecated
     public Notification() {
         this.throwable = null;
         this.value = null;

rxjava-core/src/main/java/rx/Observer.java

@@ -72,7 +72,7 @@ protected Observer(Observer<?> op) {
      * 
      * @param args
      */
-    public abstract void onNext(T args);
+    public abstract void onNext(T t);
 
     /**
      * Used to register an unsubscribe callback.

rxjava-core/src/main/java/rx/observers/EmptyObserver.java

@@ -0,0 +1,25 @@
+package rx.observers;
+
+import rx.Observer;
+
+/**
+ * Observer that does nothing... including swallowing errors.
+ */
+public class EmptyObserver<T> extends Observer<T> {
+
+    @Override
+    public void onCompleted() {
+
+    }
+
+    @Override
+    public void onError(Throwable e) {
+
+    }
+
+    @Override
+    public void onNext(T args) {
+
+    }
+
+}

rxjava-core/src/main/java/rx/observers/TestObserver.java

@@ -0,0 +1,115 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.observers;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import rx.Notification;
+import rx.Observer;
+
+/**
+ * Observer usable for unit testing to perform assertions, inspect received events or wrap a mocked Observer.
+ */
+public class TestObserver<T> extends Observer<T> {
+
+    private final Observer<Object> EMPTY = new EmptyObserver<Object>();
+
+    private final Observer<T> delegate;
+    private final ArrayList<T> onNextEvents = new ArrayList<T>();
+    private final ArrayList<Throwable> onErrorEvents = new ArrayList<Throwable>();
+    private final ArrayList<Notification<T>> onCompletedEvents = new ArrayList<Notification<T>>();
+
+    public TestObserver(Observer<T> delegate) {
+        this.delegate = delegate;
+    }
+
+    @SuppressWarnings("unchecked")
+    public TestObserver() {
+        this.delegate = (Observer<T>) EMPTY;
+    }
+
+    @Override
+    public void onCompleted() {
+        onCompletedEvents.add(Notification.<T> createOnCompleted());
+        delegate.onCompleted();
+    }
+
+    public List<Notification<T>> getOnCompletedEvents() {
+        return Collections.unmodifiableList(onCompletedEvents);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        onErrorEvents.add(e);
+        delegate.onError(e);
+    }
+
+    public List<Throwable> getOnErrorEvents() {
+        return Collections.unmodifiableList(onErrorEvents);
+    }
+
+    @Override
+    public void onNext(T t) {
+        onNextEvents.add(t);
+        delegate.onNext(t);
+    }
+
+    public List<T> getOnNextEvents() {
+        return Collections.unmodifiableList(onNextEvents);
+    }
+
+    public void assertReceivedOnNext(List<T> items) {
+        if (onNextEvents.size() != items.size()) {
+            throw new AssertionError("Number of items does not match. Provided: " + items.size() + "  Actual: " + onNextEvents.size());
+        }
+
+        for (int i = 0; i < items.size(); i++) {
+            if (items.get(i) == null) {
+                // check for null equality
+                if (onNextEvents.get(i) != null) {
+                    throw new AssertionError("Value at index: " + i + " expected to be [null] but was: [" + onNextEvents.get(i) + "]");
+                }
+            } else if (!items.get(i).equals(onNextEvents.get(i))) {
+                throw new AssertionError("Value at index: " + i + " expected to be [" + items.get(i) + "] but was: [" + onNextEvents.get(i) + "]");
+            }
+        }
+
+    }
+
+    /**
+     * Assert that a single terminal event occurred, either onCompleted or onError.
+     */
+    public void assertTerminalEvent() {
+        if (onErrorEvents.size() > 1) {
+            throw new AssertionError("Too many onError events: " + onErrorEvents.size());
+        }
+
+        if (onCompletedEvents.size() > 1) {
+            throw new AssertionError("Too many onCompleted events: " + onCompletedEvents.size());
+        }
+
+        if (onCompletedEvents.size() == 1 && onErrorEvents.size() == 1) {
+            throw new AssertionError("Received both an onError and onCompleted. Should be one or the other.");
+        }
+
+        if (onCompletedEvents.size() == 0 && onErrorEvents.size() == 0) {
+            throw new AssertionError("No terminal events received.");
+        }
+    }
+
+}

rxjava-core/src/test/java/rx/observers/TestObserverTest.java

@@ -0,0 +1,121 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.observers;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.util.Arrays;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.InOrder;
+
+import rx.Observable;
+import rx.Observer;
+import rx.subjects.PublishSubject;
+
+public class TestObserverTest {
+
+    @Rule
+    public ExpectedException thrown = ExpectedException.none();
+
+    @Test
+    public void testAssert() {
+        Observable<Integer> oi = Observable.from(Arrays.asList(1, 2));
+        TestObserver<Integer> o = new TestObserver<Integer>();
+        oi.subscribe(o);
+
+        o.assertReceivedOnNext(Arrays.asList(1, 2));
+        assertEquals(2, o.getOnNextEvents().size());
+        o.assertTerminalEvent();
+    }
+
+    @Test
+    public void testAssertNotMatchCount() {
+        Observable<Integer> oi = Observable.from(Arrays.asList(1, 2));
+        TestObserver<Integer> o = new TestObserver<Integer>();
+        oi.subscribe(o);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Number of items does not match. Provided: 1  Actual: 2");
+
+        o.assertReceivedOnNext(Arrays.asList(1));
+        assertEquals(2, o.getOnNextEvents().size());
+        o.assertTerminalEvent();
+    }
+
+    @Test
+    public void testAssertNotMatchValue() {
+        Observable<Integer> oi = Observable.from(Arrays.asList(1, 2));
+        TestObserver<Integer> o = new TestObserver<Integer>();
+        oi.subscribe(o);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value at index: 1 expected to be [3] but was: [2]");
+
+        o.assertReceivedOnNext(Arrays.asList(1, 3));
+        assertEquals(2, o.getOnNextEvents().size());
+        o.assertTerminalEvent();
+    }
+
+    @Test
+    public void testAssertTerminalEventNotReceived() {
+        PublishSubject<Integer> p = PublishSubject.create();
+        TestObserver<Integer> o = new TestObserver<Integer>();
+        p.toObservable().subscribe(o);
+
+        p.onNext(1);
+        p.onNext(2);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("No terminal events received.");
+
+        o.assertReceivedOnNext(Arrays.asList(1, 2));
+        assertEquals(2, o.getOnNextEvents().size());
+        o.assertTerminalEvent();
+    }
+
+    @Test
+    public void testWrappingMock() {
+        Observable<Integer> oi = Observable.from(Arrays.asList(1, 2));
+        @SuppressWarnings("unchecked")
+        Observer<Integer> mockObserver = mock(Observer.class);
+        oi.subscribe(new TestObserver<Integer>(mockObserver));
+
+        InOrder inOrder = inOrder(mockObserver);
+        inOrder.verify(mockObserver, times(1)).onNext(1);
+        inOrder.verify(mockObserver, times(1)).onNext(2);
+        inOrder.verify(mockObserver, times(1)).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+    }
+
+    @Test
+    public void testWrappingMockWhenUnsubscribeInvolved() {
+        Observable<Integer> oi = Observable.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9)).take(2);
+        @SuppressWarnings("unchecked")
+        Observer<Integer> mockObserver = mock(Observer.class);
+        oi.subscribe(new TestObserver<Integer>(mockObserver));
+
+        InOrder inOrder = inOrder(mockObserver);
+        inOrder.verify(mockObserver, times(1)).onNext(1);
+        inOrder.verify(mockObserver, times(1)).onNext(2);
+        inOrder.verify(mockObserver, times(1)).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+    }
+
+}