3.x: Add missing coverage, fix unused/inconsistent ops (#6901)

* 3.x: Add missing coverage, fix unused/inconsistent ops

* More coverage improvements and cleanup

* Some more coverage

* Observable coverage and cleanup

* Improve Flowable internals and coverage

* More Flowable operator coverage and fixes

* Last set of coverage & cleanup for Flowable operators

* Fix wrong use of j.u.Observable

Author: akarnokd

src/main/java/io/reactivex/rxjava3/core/Completable.java

@@ -981,7 +981,7 @@ private static Completable merge0(@NonNull Publisher<@NonNull ? extends Completa
     @SafeVarargs
     public static Completable mergeArrayDelayError(@NonNull CompletableSource... sources) {
         Objects.requireNonNull(sources, "sources is null");
-        return RxJavaPlugins.onAssembly(new CompletableMergeDelayErrorArray(sources));
+        return RxJavaPlugins.onAssembly(new CompletableMergeArrayDelayError(sources));
     }
 
     /**

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

@@ -281,13 +281,8 @@ public static <T> Predicate<T> equalsWith(T value) {
         return new EqualsPredicate<>(value);
     }
 
-    enum HashSetCallable implements Supplier<Set<Object>>, Callable<Set<Object>> {
+    enum HashSetSupplier implements Supplier<Set<Object>> {
         INSTANCE;
-        @Override
-        public Set<Object> call() {
-            return new HashSet<>();
-        }
-
         @Override
         public Set<Object> get() {
             return new HashSet<>();
@@ -296,7 +291,7 @@ public Set<Object> get() {
 
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public static <T> Supplier<Set<T>> createHashSet() {
-        return (Supplier)HashSetCallable.INSTANCE;
+        return (Supplier)HashSetSupplier.INSTANCE;
     }
 
     static final class NotificationOnNext<T> implements Consumer<T> {
@@ -742,12 +737,7 @@ public boolean test(Object o) {
         }
     }
 
-    static final class NullProvider implements Callable<Object>, Supplier<Object> {
-        @Override
-        public Object call() {
-            return null;
-        }
-
+    static final class NullProvider implements Supplier<Object> {
         @Override
         public Object get() {
             return null;

src/main/java/io/reactivex/rxjava3/internal/observers/FutureObserver.java

@@ -128,22 +128,15 @@ public void onNext(T t) {
     @Override
     public void onError(Throwable t) {
         if (error == null) {
-            error = t;
-
-            for (;;) {
-                Disposable a = upstream.get();
-                if (a == this || a == DisposableHelper.DISPOSED) {
-                    RxJavaPlugins.onError(t);
-                    return;
-                }
-                if (upstream.compareAndSet(a, this)) {
-                    countDown();
-                    return;
-                }
+            Disposable a = upstream.get();
+            if (a != this && a != DisposableHelper.DISPOSED
+                    && upstream.compareAndSet(a, this)) {
+                error = t;
+                countDown();
+                return;
             }
-        } else {
-            RxJavaPlugins.onError(t);
         }
+        RxJavaPlugins.onError(t);
     }
 
     @Override
@@ -152,15 +145,12 @@ public void onComplete() {
             onError(new NoSuchElementException("The source is empty"));
             return;
         }
-        for (;;) {
-            Disposable a = upstream.get();
-            if (a == this || a == DisposableHelper.DISPOSED) {
-                return;
-            }
-            if (upstream.compareAndSet(a, this)) {
-                countDown();
-                return;
-            }
+        Disposable a = upstream.get();
+        if (a == this || a == DisposableHelper.DISPOSED) {
+            return;
+        }
+        if (upstream.compareAndSet(a, this)) {
+            countDown();
         }
     }
 

src/main/java/io/reactivex/rxjava3/internal/observers/InnerQueuedObserver.java

@@ -114,8 +114,4 @@ public void setDone() {
     public SimpleQueue<T> queue() {
         return queue;
     }
-
-    public int fusionMode() {
-        return fusionMode;
-    }
 }

src/main/java/io/reactivex/rxjava3/internal/observers/QueueDrainObserver.java

@@ -57,10 +57,6 @@ public final boolean enter() {
         return wip.getAndIncrement() == 0;
     }
 
-    public final boolean fastEnter() {
-        return wip.get() == 0 && wip.compareAndSet(0, 1);
-    }
-
     protected final void fastPathEmit(U value, boolean delayError, Disposable dispose) {
         final Observer<? super V> observer = downstream;
         final SimplePlainQueue<U> q = queue;

src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java

@@ -182,9 +182,8 @@ void drain() {
                     boolean empty = cs == null;
 
                     if (d && empty) {
-                        if (once.compareAndSet(false, true)) {
-                            downstream.onComplete();
-                        }
+                        // errors never set done or call drain.
+                        downstream.onComplete();
                         return;
                     }
 

src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableMergeArray.java

@@ -86,9 +86,8 @@ public void onError(Throwable e) {
         @Override
         public void onComplete() {
             if (decrementAndGet() == 0) {
-                if (once.compareAndSet(false, true)) {
-                    downstream.onComplete();
-                }
+                // errors don't decrement this
+                downstream.onComplete();
             }
         }
 

src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableMergeDelayErrorArray.java renamed to src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableMergeArrayDelayError.java

@@ -19,11 +19,11 @@
 import io.reactivex.rxjava3.disposables.*;
 import io.reactivex.rxjava3.internal.util.AtomicThrowable;
 
-public final class CompletableMergeDelayErrorArray extends Completable {
+public final class CompletableMergeArrayDelayError extends Completable {
 
     final CompletableSource[] sources;
 
-    public CompletableMergeDelayErrorArray(CompletableSource[] sources) {
+    public CompletableMergeArrayDelayError(CompletableSource[] sources) {
         this.sources = sources;
     }
 

src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableMergeDelayErrorIterable.java

@@ -20,7 +20,7 @@
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
 import io.reactivex.rxjava3.exceptions.Exceptions;
-import io.reactivex.rxjava3.internal.operators.completable.CompletableMergeDelayErrorArray.*;
+import io.reactivex.rxjava3.internal.operators.completable.CompletableMergeArrayDelayError.*;
 import io.reactivex.rxjava3.internal.util.AtomicThrowable;
 
 public final class CompletableMergeDelayErrorIterable extends Completable {

src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableMergeIterable.java

@@ -128,9 +128,8 @@ public void onError(Throwable e) {
         @Override
         public void onComplete() {
             if (wip.decrementAndGet() == 0) {
-                if (compareAndSet(false, true)) {
-                    downstream.onComplete();
-                }
+                // errors don't decrement wip
+                downstream.onComplete();
             }
         }