Merge pull request #1742 from benjchristensen/issue-1571

EmptyObserver and TestObserver

src/main/java/rx/internal/operators/BufferUntilSubscriber.java

@@ -21,8 +21,7 @@
 import rx.Observer;
 import rx.Subscriber;
 import rx.functions.Action0;
-import rx.observers.EmptyObserver;
-import rx.observers.Subscribers;
+import rx.observers.Observers;
 import rx.subjects.Subject;
 import rx.subscriptions.Subscriptions;
 
@@ -51,9 +50,6 @@
  */
 public class BufferUntilSubscriber<T> extends Subject<T, T> {
 
-    @SuppressWarnings("rawtypes")
-    private final static Observer EMPTY_OBSERVER = new EmptyObserver();
-
     /**
      * @warn create() undescribed
      * @return
@@ -96,7 +92,7 @@ public void call(final Subscriber<? super T> s) {
                 s.add(Subscriptions.create(new Action0() {
                     @Override
                     public void call() {
-                        state.observerRef = EMPTY_OBSERVER;
+                        state.observerRef = Observers.empty();
                     }
                 }));
                 boolean win = false;

src/main/java/rx/observers/Observers.java

@@ -46,9 +46,7 @@ public final void onNext(Object args) {
 
     /**
      * Returns an inert {@link Observer} that does nothing in response to the emissions or notifications from
-     * any {@code Observable} it subscribes to. This is different, however, from an {@link EmptyObserver}, in
-     * that it will throw an exception if its {@link Observer#onError onError} method is called (whereas
-     * {@code EmptyObserver} will swallow the error in such a case).
+     * any {@code Observable} it subscribes to but will throw an exception if its {@link Observer#onError onError} method is called.
      *
      * @return an inert {@code Observer}
      */
@@ -59,8 +57,8 @@ public static <T> Observer<T> empty() {
 
     /**
      * Creates an {@link Observer} that receives the emissions of any {@code Observable} it subscribes to via
-     * {@link Observer#onNext onNext} but ignores {@link Observer#onError onError} and
-     * {@link Observer#onCompleted onCompleted} notifications.
+     * {@link Observer#onNext onNext} but ignores {@link Observer#onCompleted onCompleted} notifications. 
+     * It will throws an {@link OnErrorNotImplementedException} if {@link Observer#onError onError} is invoked.
      *
      * @param onNext
      *          a function that handles each item emitted by an {@code Observable}

src/main/java/rx/observers/TestObserver.java

@@ -36,8 +36,9 @@ public TestObserver(Observer<T> delegate) {
         this.delegate = delegate;
     }
 
+    @SuppressWarnings("unchecked")
     public TestObserver() {
-        this.delegate = Observers.empty();
+        this.delegate = (Observer<T>) INERT;
     }
 
     @Override
@@ -153,4 +154,23 @@ public void assertTerminalEvent() {
         }
     }
 
+    // do nothing ... including swallowing errors
+    private static Observer<Object> INERT = new Observer<Object>() {
+
+        @Override
+        public void onCompleted() {
+
+        }
+
+        @Override
+        public void onError(Throwable e) {
+
+        }
+
+        @Override
+        public void onNext(Object t) {
+
+        }
+
+    };
 }

src/test/java/rx/observers/TestObserverTest.java

@@ -119,5 +119,10 @@ public void testWrappingMockWhenUnsubscribeInvolved() {
         inOrder.verify(mockObserver, times(1)).onCompleted();
         inOrder.verifyNoMoreInteractions();
     }
+
+    @Test
+    public void testErrorSwallowed() {
+        Observable.error(new RuntimeException()).subscribe(new TestObserver<Object>());
+    }
 
 }