2.x: add missing ops, cleanup, fusion fixes (#4403)

* 2.x: add missing ops, cleanup, fusion fixes

* Fix scalar handling in flatMaps, rename Objects to ObjectHelper

Author: akarnokd

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

@@ -13,7 +13,7 @@
 
 package io.reactivex;
 
-import io.reactivex.internal.functions.Objects;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.util.NotificationLite;
 
 /**
@@ -92,7 +92,7 @@ public Throwable getError() {
     public boolean equals(Object obj) {
         if (obj instanceof Notification) {
             Notification<?> n = (Notification<?>) obj;
-            return Objects.equals(value, n.value);
+            return ObjectHelper.equals(value, n.value);
         }
         return false;
     }
@@ -123,7 +123,7 @@ public String toString() {
      * @throws NullPointerException if value is null
      */
     public static <T> Notification<T> createOnNext(T value) {
-        Objects.requireNonNull(value, "value is null");
+        ObjectHelper.requireNonNull(value, "value is null");
         return new Notification<T>(value);
     }
 
@@ -135,7 +135,7 @@ public static <T> Notification<T> createOnNext(T value) {
      * @throws NullPointerException if error is null
      */
     public static <T> Notification<T> createOnError(Throwable error) {
-        Objects.requireNonNull(error, "error is null");
+        ObjectHelper.requireNonNull(error, "error is null");
         return new Notification<T>(NotificationLite.error(error));
     }
 

src/main/java/io/reactivex/Flowable.java

@@ -16,7 +16,7 @@
 
 import io.reactivex.exceptions.*;
 import io.reactivex.internal.disposables.DisposableContainer;
-import io.reactivex.internal.functions.Objects;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.util.OpenHashSet;
 
 /**
@@ -40,10 +40,10 @@ public CompositeDisposable() {
      * @param resources the array of Disposables to start with
      */
     public CompositeDisposable(Disposable... resources) {
-        Objects.requireNonNull(resources, "resources is null");
+        ObjectHelper.requireNonNull(resources, "resources is null");
         this.resources = new OpenHashSet<Disposable>(resources.length + 1);
         for (Disposable d : resources) {
-            Objects.requireNonNull(d, "Disposable item is null");
+            ObjectHelper.requireNonNull(d, "Disposable item is null");
             this.resources.add(d);
         }
     }
@@ -53,9 +53,9 @@ public CompositeDisposable(Disposable... resources) {
      * @param resources the Iterable sequence of Disposables to start with
      */
     public CompositeDisposable(Iterable<? extends Disposable> resources) {
-        Objects.requireNonNull(resources, "resources is null");
+        ObjectHelper.requireNonNull(resources, "resources is null");
         for (Disposable d : resources) {
-            Objects.requireNonNull(d, "Disposable item is null");
+            ObjectHelper.requireNonNull(d, "Disposable item is null");
             this.resources.add(d);
         }
     }
@@ -85,7 +85,7 @@ public boolean isDisposed() {
 
     @Override
     public boolean add(Disposable d) {
-        Objects.requireNonNull(d, "d is null");
+        ObjectHelper.requireNonNull(d, "d is null");
         if (!disposed) {
             synchronized (this) {
                 if (!disposed) {
@@ -110,7 +110,7 @@ public boolean add(Disposable d) {
      * @return true if the operation was successful, false if the container has been disposed
      */
     public boolean addAll(Disposable... ds) {
-        Objects.requireNonNull(ds, "ds is null");
+        ObjectHelper.requireNonNull(ds, "ds is null");
         if (!disposed) {
             synchronized (this) {
                 if (!disposed) {
@@ -120,7 +120,7 @@ public boolean addAll(Disposable... ds) {
                         resources = set;
                     }
                     for (Disposable d : ds) {
-                        Objects.requireNonNull(d, "d is null");
+                        ObjectHelper.requireNonNull(d, "d is null");
                         set.add(d);
                     }
                     return true;
@@ -144,7 +144,7 @@ public boolean remove(Disposable d) {
 
     @Override
     public boolean delete(Disposable d) {
-        Objects.requireNonNull(d, "Disposable item is null");
+        ObjectHelper.requireNonNull(d, "Disposable item is null");
         if (disposed) {
             return false;
         }

src/main/java/io/reactivex/Notification.java

@@ -39,7 +39,7 @@ private Disposables() {
      * @return the new Disposable instance
      */
     public static Disposable from(Runnable run) {
-        Objects.requireNonNull(run, "run is null");
+        ObjectHelper.requireNonNull(run, "run is null");
         return new RunnableDisposable(run);
     }
 
@@ -50,7 +50,7 @@ public static Disposable from(Runnable run) {
      * @return the new Disposable instance
      */
     public static Disposable from(Action run) {
-        Objects.requireNonNull(run, "run is null");
+        ObjectHelper.requireNonNull(run, "run is null");
         return new ActionDisposable(run);
     }
 
@@ -61,7 +61,7 @@ public static Disposable from(Action run) {
      * @return the new Disposable instance
      */
     public static Disposable from(Future<?> future) {
-        Objects.requireNonNull(future, "future is null");
+        ObjectHelper.requireNonNull(future, "future is null");
         return from(future, true);
     }
 
@@ -73,7 +73,7 @@ public static Disposable from(Future<?> future) {
      * @return the new Disposable instance
      */
     public static Disposable from(Future<?> future, boolean allowInterrupt) {
-        Objects.requireNonNull(future, "future is null");
+        ObjectHelper.requireNonNull(future, "future is null");
         return new FutureDisposable(future, allowInterrupt);
     }
 
@@ -84,7 +84,7 @@ public static Disposable from(Future<?> future, boolean allowInterrupt) {
      * @return the new Disposable instance
      */
     public static Disposable from(Subscription subscription) {
-        Objects.requireNonNull(subscription, "subscription is null");
+        ObjectHelper.requireNonNull(subscription, "subscription is null");
         return new SubscriptionDisposable(subscription);
     }
 

src/main/java/io/reactivex/Observable.java

@@ -15,7 +15,7 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
-import io.reactivex.internal.functions.Objects;
+import io.reactivex.internal.functions.ObjectHelper;
 
 /**
  * Base class for Disposable containers that manage some other type that
@@ -28,7 +28,7 @@ abstract class ReferenceDisposable<T> extends AtomicReference<T> implements Disp
     private static final long serialVersionUID = 6537757548749041217L;
 
     ReferenceDisposable(T value) {
-        super(Objects.requireNonNull(value, "value is null"));
+        super(ObjectHelper.requireNonNull(value, "value is null"));
     }
 
     protected abstract void onDisposed(T value);

src/main/java/io/reactivex/Single.java

@@ -16,7 +16,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.functions.Objects;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -54,7 +54,7 @@ public static boolean set(AtomicReference<Disposable> field, Disposable d) {
     }
 
     public static boolean setOnce(AtomicReference<Disposable> field, Disposable d) {
-        Objects.requireNonNull(d, "d is null");
+        ObjectHelper.requireNonNull(d, "d is null");
         if (!field.compareAndSet(null, d)) {
             d.dispose();
             if (field.get() != DISPOSED) {

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

@@ -14,9 +14,22 @@
 package io.reactivex.internal.disposables;
 
 import io.reactivex.*;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.fuseable.QueueDisposable;
 
-public enum EmptyDisposable implements Disposable {
+/**
+ * Represents a stateless empty Disposable that reports being always
+ * empty and disposed.
+ * <p>It is also async-fuseable but empty all the time.
+ * <p>Since EmptyDisposable implements QueueDisposable and is empty,
+ * don't use it in tests and then signal onNext with it;
+ * use Disposables.empty() instead.
+ */
+public enum EmptyDisposable implements QueueDisposable<Object> {
+    /**
+     * Since EmptyDisposable implements QueueDisposable and is empty,
+     * don't use it in tests and then signal onNext with it;
+     * use Disposables.empty() instead.
+     */
     INSTANCE
     ;
 
@@ -27,7 +40,7 @@ public void dispose() {
 
     @Override
     public boolean isDisposed() {
-        return true; // TODO is this okay?
+        return true;
     }
 
     public static void complete(Observer<?> s) {
@@ -55,4 +68,35 @@ public static void error(Throwable e, SingleObserver<?> s) {
         s.onError(e);
     }
 
+    @Override
+    public boolean offer(Object value) {
+        throw new UnsupportedOperationException("Should not be called!");
+    }
+
+    @Override
+    public boolean offer(Object v1, Object v2) {
+        throw new UnsupportedOperationException("Should not be called!");
+    }
+
+    @Override
+    public Object poll() throws Exception {
+        return null; // always empty
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return true; // always empty
+    }
+
+    @Override
+    public void clear() {
+        // nothing to do
+    }
+
+    @Override
+    public int requestFusion(int mode) {
+        return mode & ASYNC;
+    }
+    
+
 }

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

@@ -16,7 +16,7 @@
 
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.*;
-import io.reactivex.internal.functions.Objects;
+import io.reactivex.internal.functions.ObjectHelper;
 
 /**
  * A disposable container that can hold onto multiple other disposables.
@@ -31,18 +31,18 @@ public ListCompositeDisposable() {
     }
 
     public ListCompositeDisposable(Disposable... resources) {
-        Objects.requireNonNull(resources, "resources is null");
+        ObjectHelper.requireNonNull(resources, "resources is null");
         this.resources = new LinkedList<Disposable>();
         for (Disposable d : resources) {
-            Objects.requireNonNull(d, "Disposable item is null");
+            ObjectHelper.requireNonNull(d, "Disposable item is null");
             this.resources.add(d);
         }
     }
 
     public ListCompositeDisposable(Iterable<? extends Disposable> resources) {
-        Objects.requireNonNull(resources, "resources is null");
+        ObjectHelper.requireNonNull(resources, "resources is null");
         for (Disposable d : resources) {
-            Objects.requireNonNull(d, "Disposable item is null");
+            ObjectHelper.requireNonNull(d, "Disposable item is null");
             this.resources.add(d);
         }
     }
@@ -72,7 +72,7 @@ public boolean isDisposed() {
 
     @Override
     public boolean add(Disposable d) {
-        Objects.requireNonNull(d, "d is null");
+        ObjectHelper.requireNonNull(d, "d is null");
         if (!disposed) {
             synchronized (this) {
                 if (!disposed) {
@@ -91,7 +91,7 @@ public boolean add(Disposable d) {
     }
 
     public boolean addAll(Disposable... ds) {
-        Objects.requireNonNull(ds, "ds is null");
+        ObjectHelper.requireNonNull(ds, "ds is null");
         if (!disposed) {
             synchronized (this) {
                 if (!disposed) {
@@ -101,7 +101,7 @@ public boolean addAll(Disposable... ds) {
                         resources = set;
                     }
                     for (Disposable d : ds) {
-                        Objects.requireNonNull(d, "d is null");
+                        ObjectHelper.requireNonNull(d, "d is null");
                         set.add(d);
                     }
                     return true;
@@ -125,7 +125,7 @@ public boolean remove(Disposable d) {
 
     @Override
     public boolean delete(Disposable d) {
-        Objects.requireNonNull(d, "Disposable item is null");
+        ObjectHelper.requireNonNull(d, "Disposable item is null");
         if (disposed) {
             return false;
         }