2.x: cleanup, behavior clarifications, fixes, coverage 8/28-1 (#4435)

Author: akarnokd

src/main/java/io/reactivex/disposables/RunnableDisposable.java

@@ -27,4 +27,9 @@ final class RunnableDisposable extends ReferenceDisposable<Runnable> {
     protected void onDisposed(Runnable value) {
         value.run();
     }
+    
+    @Override
+    public String toString() {
+        return "RunnableDisposable(disposed=" + isDisposed() + ", " + get() + ")";
+    }
 }

src/main/java/io/reactivex/disposables/SerialDisposable.java

@@ -15,7 +15,7 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 
 /**
  * A Disposable container that allows atomically updating/replacing the contained

src/main/java/io/reactivex/internal/disposables/DisposableHelper.java

@@ -22,14 +22,13 @@
 /**
  * Utility methods for working with Disposables atomically.
  */
-public enum DisposableHelper {
-    ;
-    
-    /**
-     * Marker instance compared by identity for indicating a previously referenced
-     * {@link Disposable} was disposed. DO NOT USE this instance as an arbitrary, empty disposable!
+public enum DisposableHelper implements Disposable {
+    /** 
+     * The singleton instance representing a terminal, disposed state;
+     * Don't leak it!
      */
-    public static final Disposable DISPOSED = Disposed.INSTANCE;
+    DISPOSED
+    ;
 
     public static boolean isDisposed(Disposable d) {
         return d == DISPOSED;
@@ -82,9 +81,10 @@ public static boolean replace(AtomicReference<Disposable> field, Disposable d) {
 
     public static boolean dispose(AtomicReference<Disposable> field) {
         Disposable current = field.get();
-        if (current != DISPOSED) {
-            current = field.getAndSet(DISPOSED);
-            if (current != DISPOSED) {
+        Disposable d = DISPOSED;
+        if (current != d) {
+            current = field.getAndSet(d);
+            if (current != d) {
                 if (current != null) {
                     current.dispose();
                 }
@@ -121,18 +121,13 @@ public static void reportDisposableSet() {
         RxJavaPlugins.onError(new IllegalStateException("Disposable already set!"));
     }
 
-    enum Disposed implements Disposable {
-        INSTANCE;
-        
-        @Override
-        public void dispose() {
-            // deliberately no-op
-        }
-
-        @Override
-        public boolean isDisposed() {
-            return true;
-        }
+    @Override
+    public void dispose() {
+        // deliberately no-op
+    }
+    
+    @Override
+    public boolean isDisposed() {
+        return true;
     }
-
 }

src/main/java/io/reactivex/internal/disposables/ListCompositeDisposable.java

@@ -42,6 +42,7 @@ public ListCompositeDisposable(Disposable... resources) {
 
     public ListCompositeDisposable(Iterable<? extends Disposable> resources) {
         ObjectHelper.requireNonNull(resources, "resources is null");
+        this.resources = new LinkedList<Disposable>();
         for (Disposable d : resources) {
             ObjectHelper.requireNonNull(d, "Disposable item is null");
             this.resources.add(d);

src/main/java/io/reactivex/internal/functions/Functions.java

@@ -148,6 +148,11 @@ public R apply(Object[] a) throws Exception {
         public Object apply(Object v) {
             return v;
         }
+
+        @Override
+        public String toString() {
+            return "IdentityFunction";
+        }
     };
 
     /**
@@ -163,16 +168,31 @@ public static <T> Function<T, T> identity() {
     public static final Runnable EMPTY_RUNNABLE = new Runnable() {
         @Override
         public void run() { }
+        
+        @Override
+        public String toString() {
+            return "EmptyRunnable";
+        }
     };
 
     public static final Action EMPTY_ACTION = new Action() {
         @Override
         public void run() { }
+
+        @Override
+        public String toString() {
+            return "EmptyAction";
+        }
     };
 
     static final Consumer<Object> EMPTY_CONSUMER = new Consumer<Object>() {
         @Override
         public void accept(Object v) { }
+
+        @Override
+        public String toString() {
+            return "EmptyConsumer";
+        }
     };
 
     /**

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundarySupplier.java

@@ -21,7 +21,7 @@
 
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.flowable.QueueDrainSubscriber;
 import io.reactivex.internal.subscriptions.*;

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferTimed.java

@@ -19,11 +19,11 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.*;
+import io.reactivex.Scheduler;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.flowable.QueueDrainSubscriber;
 import io.reactivex.internal.subscriptions.*;

src/main/java/io/reactivex/internal/operators/flowable/FlowableDebounceTimed.java

@@ -18,10 +18,10 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.*;
+import io.reactivex.Scheduler;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 import io.reactivex.plugins.RxJavaPlugins;

src/main/java/io/reactivex/internal/operators/flowable/FlowableInterval.java

@@ -21,7 +21,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.MissingBackpressureException;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableIntervalRange.java

@@ -20,7 +20,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowTimed.java

@@ -24,7 +24,7 @@
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.fuseable.SimpleQueue;
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.flowable.QueueDrainSubscriber;

src/main/java/io/reactivex/internal/operators/observable/ObservableDebounce.java

@@ -19,7 +19,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Function;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 

src/main/java/io/reactivex/internal/operators/observable/ObservableDebounceTimed.java

@@ -19,7 +19,7 @@
 import io.reactivex.*;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.observers.SerializedObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 

src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMap.java

@@ -22,7 +22,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.Function;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.queue.*;
 

src/main/java/io/reactivex/internal/operators/observable/ObservableInterval.java

@@ -18,7 +18,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 
 public final class ObservableInterval extends Observable<Long> {
     final Scheduler scheduler;

src/main/java/io/reactivex/internal/operators/observable/ObservableIntervalRange.java

@@ -18,7 +18,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 
 public final class ObservableIntervalRange extends Observable<Long> {
     final Scheduler scheduler;

src/main/java/io/reactivex/internal/operators/observable/ObservableSampleWithObservable.java

@@ -17,7 +17,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.observers.SerializedObserver;
 
 public final class ObservableSampleWithObservable<T> extends AbstractObservableWithUpstream<T, T> {

src/main/java/io/reactivex/internal/operators/observable/ObservableThrottleFirstTimed.java

@@ -19,7 +19,7 @@
 import io.reactivex.*;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.observers.SerializedObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 

src/main/java/io/reactivex/internal/operators/observable/ObserverResourceWrapper.java

@@ -17,7 +17,7 @@
 
 import io.reactivex.Observer;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 
 public final class ObserverResourceWrapper<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
     /** */

src/main/java/io/reactivex/internal/operators/single/SingleTakeUntil.java

@@ -20,7 +20,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 

src/main/java/io/reactivex/internal/subscribers/completable/CallbackCompletableObserver.java

@@ -19,7 +19,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class CallbackCompletableObserver

src/main/java/io/reactivex/internal/subscribers/completable/EmptyCompletableObserver.java

@@ -17,7 +17,7 @@
 
 import io.reactivex.CompletableObserver;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class EmptyCompletableObserver

src/main/java/io/reactivex/internal/subscribers/flowable/ForEachWhileSubscriber.java

@@ -18,7 +18,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.disposables.Disposable;
-import io.reactivex.exceptions.Exceptions;
+import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -85,7 +85,7 @@ public void onError(Throwable t) {
             onError.accept(t);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
-            RxJavaPlugins.onError(ex);
+            RxJavaPlugins.onError(new CompositeException(t, ex));
         }
     }
 

src/main/java/io/reactivex/internal/subscribers/observable/BlockingObserver.java

@@ -16,9 +16,9 @@
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicReference;
 
-import io.reactivex.*;
+import io.reactivex.Observer;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.util.NotificationLite;
 
 public final class BlockingObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {