2.x: cleanup and coverage 10/03 (#4661)

Author: akarnokd

src/main/java/io/reactivex/Maybe.java

@@ -23,6 +23,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
+import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.observers.BlockingMultiObserver;
 import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.operators.maybe.*;
@@ -2866,9 +2867,13 @@ public final Completable toCompletable() {
      * </dl>
      * @return the new Flowable instance
      */
+    @SuppressWarnings("unchecked")
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Flowable<T> toFlowable() {
+        if (this instanceof FuseToFlowable) {
+            return ((FuseToFlowable<T>)this).fuseToFlowable();
+        }
         return RxJavaPlugins.onAssembly(new MaybeToFlowable<T>(this));
     }
 
@@ -2881,8 +2886,12 @@ public final Flowable<T> toFlowable() {
      * </dl>
      * @return the new Observable instance
      */
+    @SuppressWarnings("unchecked")
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Observable<T> toObservable() {
+        if (this instanceof FuseToObservable) {
+            return ((FuseToObservable<T>)this).fuseToObservable();
+        }
         return RxJavaPlugins.onAssembly(new MaybeToObservable<T>(this));
     }
 

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

@@ -30,7 +30,11 @@ public enum EmptyDisposable implements QueueDisposable<Object> {
      * don't use it in tests and then signal onNext with it;
      * use Disposables.empty() instead.
      */
-    INSTANCE
+    INSTANCE,
+    /**
+     * An empty disposable that returns false for isDisposed.
+     */
+    NEVER
     ;
 
     @Override
@@ -40,7 +44,7 @@ public void dispose() {
 
     @Override
     public boolean isDisposed() {
-        return true;
+        return this == INSTANCE;
     }
 
     public static void complete(Observer<?> s) {

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

@@ -110,33 +110,28 @@ public void onSubscribe(Disposable s) {
 
     @Override
     public void onSuccess(T t) {
-        if (value != null) {
-            s.get().dispose();
-            onError(new IndexOutOfBoundsException("More than one element received"));
+        Disposable a = s.get();
+        if (a == DisposableHelper.DISPOSED) {
             return;
         }
         value = t;
+        s.compareAndSet(a, this);
         countDown();
     }
 
     @Override
     public void onError(Throwable t) {
-        if (error == null) {
+        for (;;) {
+            Disposable a = s.get();
+            if (a == DisposableHelper.DISPOSED) {
+                RxJavaPlugins.onError(t);
+                return;
+            }
             error = t;
-
-            for (;;) {
-                Disposable a = s.get();
-                if (a == this || a == DisposableHelper.DISPOSED) {
-                    RxJavaPlugins.onError(t);
-                    return;
-                }
-                if (s.compareAndSet(a, this)) {
-                    countDown();
-                    return;
-                }
+            if (s.compareAndSet(a, this)) {
+                countDown();
+                return;
             }
-        } else {
-            RxJavaPlugins.onError(t);
         }
     }
 

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

@@ -24,7 +24,7 @@ private CompletableNever() {
 
     @Override
     protected void subscribeActual(CompletableObserver s) {
-        s.onSubscribe(EmptyDisposable.INSTANCE);
+        s.onSubscribe(EmptyDisposable.NEVER);
     }
 
 }

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

@@ -115,10 +115,6 @@ public boolean tryOnNext(T t) {
                 return false;
             }
 
-            if (sourceMode != NONE) {
-                return actual.tryOnNext(null);
-            }
-
             U v;
 
             try {

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.*;
+import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.fuseable.SimpleQueue;
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.QueueDrainSubscriber;
@@ -733,7 +733,7 @@ public void onNext(T t) {
                     return;
                 }
             } else {
-                queue.offer(NotificationLite.next(t));
+                queue.offer(t);
                 if (!enter()) {
                     return;
                 }
@@ -797,6 +797,7 @@ void complete(UnicastProcessor<T> w) {
             }
         }
 
+        @SuppressWarnings("unchecked")
         void drainLoop() {
             final SimpleQueue<Object> q = queue;
             final Subscriber<? super Flowable<T>> a = actual;
@@ -854,7 +855,6 @@ void drainLoop() {
                     }
 
                     if (sw) {
-                        @SuppressWarnings("unchecked")
                         SubjectWork<T> work = (SubjectWork<T>)v;
 
                         if (work.open) {
@@ -889,10 +889,10 @@ public void run() {
                             }
                             continue;
                         }
-                    }
-
-                    for (UnicastProcessor<T> w : ws) {
-                        w.onNext(NotificationLite.<T>getValue(v));
+                    } else {
+                        for (UnicastProcessor<T> w : ws) {
+                            w.onNext((T)v);
+                        }
                     }
                 }