2.x: coverage, fixes and cleanup 10/11-2 (#4692)

Author: akarnokd

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

@@ -0,0 +1,53 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.observers;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.reactivex.SingleObserver;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+/**
+ * A SingleObserver implementation used for subscribing to the actual SingleSource
+ * and replace the current Disposable in a parent AtomicReference.
+ *
+ * @param <T> the value type
+ */
+public final class ResumeSingleObserver<T> implements SingleObserver<T> {
+
+    final AtomicReference<Disposable> parent;
+
+    final SingleObserver<? super T> actual;
+
+    public ResumeSingleObserver(AtomicReference<Disposable> parent, SingleObserver<? super T> actual) {
+        this.parent = parent;
+        this.actual = actual;
+    }
+
+    @Override
+    public void onSubscribe(Disposable d) {
+        DisposableHelper.replace(parent, d);
+    }
+
+    @Override
+    public void onSuccess(T value) {
+        actual.onSuccess(value);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        actual.onError(e);
+    }
+}

src/main/java/io/reactivex/internal/operators/completable/CompletableDisposeOn.java

@@ -14,7 +14,9 @@
 package io.reactivex.internal.operators.completable;
 
 import io.reactivex.*;
-import io.reactivex.disposables.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class CompletableDisposeOn extends Completable {
 
@@ -29,34 +31,65 @@ public CompletableDisposeOn(CompletableSource source, Scheduler scheduler) {
 
     @Override
     protected void subscribeActual(final CompletableObserver s) {
-        source.subscribe(new CompletableObserver() {
+        source.subscribe(new CompletableObserverImplementation(s, scheduler));
+    }
+
+    static final class CompletableObserverImplementation implements CompletableObserver, Disposable, Runnable {
+        final CompletableObserver s;
+
+        final Scheduler scheduler;
+
+        Disposable d;
+
+        volatile boolean disposed;
 
-            @Override
-            public void onComplete() {
-                s.onComplete();
+        CompletableObserverImplementation(CompletableObserver s, Scheduler scheduler) {
+            this.s = s;
+            this.scheduler = scheduler;
+        }
+
+        @Override
+        public void onComplete() {
+            if (disposed) {
+                return;
             }
+            s.onComplete();
+        }
 
-            @Override
-            public void onError(Throwable e) {
-                s.onError(e);
+        @Override
+        public void onError(Throwable e) {
+            if (disposed) {
+                RxJavaPlugins.onError(e);
+                return;
             }
+            s.onError(e);
+        }
+
+        @Override
+        public void onSubscribe(final Disposable d) {
+            if (DisposableHelper.validate(this.d, d)) {
+                this.d = d;
 
-            @Override
-            public void onSubscribe(final Disposable d) {
-                s.onSubscribe(Disposables.fromRunnable(new Runnable() {
-                    @Override
-                    public void run() {
-                        scheduler.scheduleDirect(new Runnable() {
-                            @Override
-                            public void run() {
-                                d.dispose();
-                            }
-                        });
-                    }
-                }));
+                s.onSubscribe(this);
             }
+        }
+
+        @Override
+        public void dispose() {
+            disposed = true;
+            scheduler.scheduleDirect(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return disposed;
+        }
 
-        });
+        @Override
+        public void run() {
+            d.dispose();
+            d = DisposableHelper.DISPOSED;
+        }
     }
 
 }

src/main/java/io/reactivex/internal/operators/completable/CompletableDoOnEvent.java

@@ -52,7 +52,7 @@ public void onError(Throwable e) {
                     onEvent.accept(e);
                 } catch (Throwable ex) {
                     Exceptions.throwIfFatal(ex);
-                    e = new CompositeException(ex, e);
+                    e = new CompositeException(e, ex);
                 }
 
                 s.onError(e);

src/main/java/io/reactivex/internal/operators/completable/CompletableResumeNext.java

@@ -37,6 +37,7 @@ public CompletableResumeNext(CompletableSource source,
     protected void subscribeActual(final CompletableObserver s) {
 
         final SequentialDisposable sd = new SequentialDisposable();
+        s.onSubscribe(sd);
         source.subscribe(new CompletableObserver() {
 
             @Override

src/main/java/io/reactivex/internal/operators/completable/CompletableSubscribeOn.java

@@ -32,35 +32,38 @@ public CompletableSubscribeOn(CompletableSource source, Scheduler scheduler) {
     @Override
     protected void subscribeActual(final CompletableObserver s) {
 
-        final SubscribeOnObserver parent = new SubscribeOnObserver(s);
+        final SubscribeOnObserver parent = new SubscribeOnObserver(s, source);
         s.onSubscribe(parent);
 
-        Disposable f = scheduler.scheduleDirect(new Runnable() {
-            @Override
-            public void run() {
-                source.subscribe(parent);
-            }
-        });
+        Disposable f = scheduler.scheduleDirect(parent);
 
         parent.task.replace(f);
 
     }
 
     static final class SubscribeOnObserver
     extends AtomicReference<Disposable>
-    implements CompletableObserver, Disposable {
+    implements CompletableObserver, Disposable, Runnable {
 
         private static final long serialVersionUID = 7000911171163930287L;
 
         final CompletableObserver actual;
 
         final SequentialDisposable task;
 
-        SubscribeOnObserver(CompletableObserver actual) {
+        final CompletableSource source;
+
+        SubscribeOnObserver(CompletableObserver actual, CompletableSource source) {
             this.actual = actual;
+            this.source = source;
             this.task = new SequentialDisposable();
         }
 
+        @Override
+        public void run() {
+            source.subscribe(this);
+        }
+
         @Override
         public void onSubscribe(Disposable d) {
             DisposableHelper.setOnce(this, d);
@@ -84,7 +87,7 @@ public void dispose() {
 
         @Override
         public boolean isDisposed() {
-            return DisposableHelper.isDisposed(this);
+            return DisposableHelper.isDisposed(get());
         }
     }