2.x: fix CallbackCompletableObserver calling onError, ResourceX wordi…

…ng (#5240)

src/main/java/io/reactivex/internal/observers/CallbackCompletableObserver.java

@@ -52,8 +52,7 @@ public void onComplete() {
             onComplete.run();
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
-            onError(ex);
-            return;
+            RxJavaPlugins.onError(ex);
         }
         lazySet(DisposableHelper.DISPOSED);
     }

src/main/java/io/reactivex/observers/DisposableObserver.java

@@ -24,7 +24,7 @@
  *
  * <p>All pre-implemented final methods are thread-safe.
  *
- * <p>Use the protected {@link #dispose()} to dispose the sequence from within an
+ * <p>Use the public {@link #dispose()} method to dispose the sequence from within an
  * {@code onNext} implementation.
  *
  * <p>Like all other consumers, {@code DefaultObserver} can be subscribed only once.

src/main/java/io/reactivex/observers/ResourceCompletableObserver.java

@@ -29,7 +29,7 @@
  * <p>Override the protected {@link #onStart()} to perform initialization when this
  * {@code ResourceCompletableObserver} is subscribed to a source.
  *
- * <p>Use the protected {@link #dispose()} to dispose the sequence externally and release
+ * <p>Use the public {@link #dispose()} method to dispose the sequence externally and release
  * all resources.
  *
  * <p>To release the associated resources, one has to call {@link #dispose()}

src/main/java/io/reactivex/observers/ResourceMaybeObserver.java

@@ -33,7 +33,7 @@
  * <p>Override the protected {@link #onStart()} to perform initialization when this
  * {@code ResourceMaybeObserver} is subscribed to a source.
  *
- * <p>Use the protected {@link #dispose()} to dispose the sequence externally and release
+ * <p>Use the public {@link #dispose()} method to dispose the sequence externally and release
  * all resources.
  *
  * <p>To release the associated resources, one has to call {@link #dispose()}

src/main/java/io/reactivex/observers/ResourceSingleObserver.java

@@ -30,7 +30,7 @@
  * <p>Override the protected {@link #onStart()} to perform initialization when this
  * {@code ResourceSingleObserver} is subscribed to a source.
  *
- * <p>Use the protected {@link #dispose()} to dispose the sequence externally and release
+ * <p>Use the public {@link #dispose()} method to dispose the sequence externally and release
  * all resources.
  *
  * <p>To release the associated resources, one has to call {@link #dispose()}

src/test/java/io/reactivex/completable/CompletableTest.java

@@ -2566,18 +2566,24 @@ public void run() { }
 
     @Test(timeout = 5000)
     public void subscribeTwoCallbacksCompleteThrows() {
-        final AtomicReference<Throwable> err = new AtomicReference<Throwable>();
-        normal.completable.subscribe(new Action() {
-            @Override
-            public void run() { throw new TestException(); }
-        }, new Consumer<Throwable>() {
-            @Override
-            public void accept(Throwable e) {
-                err.set(e);
-            }
-        });
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final AtomicReference<Throwable> err = new AtomicReference<Throwable>();
+            normal.completable.subscribe(new Action() {
+                @Override
+                public void run() { throw new TestException(); }
+            }, new Consumer<Throwable>() {
+                @Override
+                public void accept(Throwable e) {
+                    err.set(e);
+                }
+            });
 
-        Assert.assertTrue(String.valueOf(err.get()), err.get() instanceof TestException);
+            Assert.assertNull(err.get());
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
     }
 
     @Test(timeout = 5000)

src/test/java/io/reactivex/schedulers/TimedTest.java

@@ -84,7 +84,7 @@ public void toStringOf() {
 
         assertEquals("Timed[time=5, unit=SECONDS, value=1]", t1.toString());
     }
-    
+
     @Test(expected = NullPointerException.class)
     public void timeUnitNullFail() throws Exception {
         new Timed<Integer>(1, 5, null);