Normalize empty and never sources. (ReactiveX#4316)

JakeWharton · akarnokd · commit f8694361a7bb · 2016-08-08T18:01:28.000+02:00
diff --git a/src/main/java/io/reactivex/Completable.java b/src/main/java/io/reactivex/Completable.java
@@ -32,12 +32,6 @@
  * The class follows a similar event pattern as Reactive-Streams: onSubscribe (onError|onComplete)?
  */
 public abstract class Completable implements CompletableSource {
-    /** Single instance of a complete Completable. */
-    static final Completable COMPLETE = new CompletableEmpty();
-    
-    /** Single instance of a never Completable. */
-    static final Completable NEVER = new CompletableNever();
-
     /**
      * Wraps the given CompletableSource into a Completable
      * if not already Completable.
@@ -93,7 +87,7 @@ public static Completable amb(final Iterable<? extends CompletableSource> source
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public static Completable complete() {
-        return COMPLETE;
+        return CompletableEmpty.INSTANCE;
     }
     
     /**
@@ -432,7 +426,7 @@ public static Completable mergeDelayError(Publisher<? extends CompletableSource>
      * @return the singleton instance that never calls onError or onComplete
      */
     public static Completable never() {
-        return NEVER;
+        return CompletableNever.INSTANCE;
     }
     
     /**
diff --git a/src/main/java/io/reactivex/Flowable.java b/src/main/java/io/reactivex/Flowable.java
@@ -31,7 +31,6 @@
 import io.reactivex.internal.operators.flowable.FlowableConcatMap.ErrorMode;
 import io.reactivex.internal.schedulers.ImmediateThinScheduler;
 import io.reactivex.internal.subscribers.flowable.*;
-import io.reactivex.internal.subscriptions.EmptySubscription;
 import io.reactivex.internal.util.ArrayListSupplier;
 import io.reactivex.internal.util.HashMapSupplier;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -46,14 +45,6 @@ public abstract class Flowable<T> implements Publisher<T> {
         BUFFER_SIZE = Math.max(16, Integer.getInteger("rx2.buffer-size", 128));
     }
 
-    /** A never observable instance as there is no need to instantiate this more than once. */
-    static final Flowable<Object> NEVER = new Flowable<Object>() { // FIXME factor out
-        @Override
-        public void subscribeActual(Subscriber<? super Object> s) {
-            s.onSubscribe(EmptySubscription.INSTANCE);
-        }
-    };
-
     public static <T> Flowable<T> amb(Iterable<? extends Publisher<? extends T>> sources) {
         Objects.requireNonNull(sources, "sources is null");
         return new FlowableAmb<T>(null, sources);
@@ -480,8 +471,9 @@ public static <T> Flowable<T> defer(Callable<? extends Publisher<? extends T>> s
 
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
     @SchedulerSupport(SchedulerSupport.NONE)
+    @SuppressWarnings("unchecked")
     public static <T> Flowable<T> empty() {
-        return FlowableEmpty.empty();
+        return (Flowable<T>) FlowableEmpty.INSTANCE;
     }
 
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
@@ -1014,7 +1006,7 @@ public static <T> Flowable<T> mergeDelayError(
     @SchedulerSupport(SchedulerSupport.NONE)
     @SuppressWarnings("unchecked")
     public static <T> Flowable<T> never() {
-        return (Flowable<T>)NEVER;
+        return (Flowable<T>) FlowableNever.INSTANCE;
     }
 
     @BackpressureSupport(BackpressureKind.FULL)
diff --git a/src/main/java/io/reactivex/Single.java b/src/main/java/io/reactivex/Single.java
@@ -34,9 +34,6 @@
  * @param <T> the value type
  */
 public abstract class Single<T> implements SingleSource<T> {
-
-    static final Single<Object> NEVER = new SingleNever<Object>();
-
     static <T> Single<T> wrap(SingleSource<T> source) {
         Objects.requireNonNull(source, "source is null");
         if (source instanceof Single) {
@@ -389,7 +386,7 @@ public static <T> Flowable<T> merge(
     
     @SuppressWarnings("unchecked")
     public static <T> Single<T> never() {
-        return (Single<T>)NEVER; 
+        return (Single<T>) SingleNever.INSTANCE;
     }
     
     public static Single<Long> timer(long delay, TimeUnit unit) {
diff --git a/src/main/java/io/reactivex/internal/operators/completable/CompletableEmpty.java b/src/main/java/io/reactivex/internal/operators/completable/CompletableEmpty.java
@@ -17,6 +17,11 @@
 import io.reactivex.internal.disposables.EmptyDisposable;
 
 public final class CompletableEmpty extends Completable {
+    public static final Completable INSTANCE = new CompletableEmpty();
+
+    private CompletableEmpty() {
+    }
+
     @Override
     public void subscribeActual(CompletableObserver s) {
         s.onSubscribe(EmptyDisposable.INSTANCE);
diff --git a/src/main/java/io/reactivex/internal/operators/completable/CompletableNever.java b/src/main/java/io/reactivex/internal/operators/completable/CompletableNever.java
@@ -17,6 +17,10 @@
 import io.reactivex.internal.disposables.EmptyDisposable;
 
 public final class CompletableNever extends Completable {
+    public static final Completable INSTANCE = new CompletableNever();
+
+    private CompletableNever() {
+    }
 
     @Override
     protected void subscribeActual(CompletableObserver s) {
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableEmpty.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableEmpty.java
@@ -24,11 +24,9 @@
  */
 public final class FlowableEmpty extends Flowable<Object> implements ScalarCallable<Object> {
     
-    static final Flowable<Object> INSTANCE = new FlowableEmpty();
+    public static final Flowable<Object> INSTANCE = new FlowableEmpty();
     
-    @SuppressWarnings("unchecked")
-    public static <T> Flowable<T> empty() {
-        return (Flowable<T>)INSTANCE;
+    private FlowableEmpty() {
     }
     
     @Override
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableNever.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableNever.java
@@ -0,0 +1,29 @@
+/**
+ * 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.operators.flowable;
+
+import io.reactivex.Flowable;
+import io.reactivex.internal.subscriptions.EmptySubscription;
+import org.reactivestreams.Subscriber;
+
+public final class FlowableNever extends Flowable<Object> {
+    public static final Flowable<Object> INSTANCE = new FlowableNever();
+
+    private FlowableNever() {
+    }
+
+    @Override
+    public void subscribeActual(Subscriber<? super Object> s) {
+        s.onSubscribe(EmptySubscription.INSTANCE);
+    }
+}
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableEmpty.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableEmpty.java
@@ -15,8 +15,9 @@
 import io.reactivex.Observable;
 import io.reactivex.Observer;
 import io.reactivex.internal.disposables.EmptyDisposable;
+import io.reactivex.internal.fuseable.ScalarCallable;
 
-public final class ObservableEmpty extends Observable<Object> {
+public final class ObservableEmpty extends Observable<Object> implements ScalarCallable<Object> {
     public static final Observable<Object> INSTANCE = new ObservableEmpty();
 
     private ObservableEmpty() {
@@ -27,4 +28,9 @@ protected void subscribeActual(Observer<? super Object> o) {
         o.onSubscribe(EmptyDisposable.INSTANCE);
         o.onComplete();
     }
+
+    @Override
+    public Object call() {
+        return null; // null scalar is interpreted as being empty
+    }
 }
diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleNever.java b/src/main/java/io/reactivex/internal/operators/single/SingleNever.java
@@ -16,10 +16,14 @@
 import io.reactivex.*;
 import io.reactivex.internal.disposables.EmptyDisposable;
 
-public final class SingleNever<T> extends Single<T> {
+public final class SingleNever extends Single<Object> {
+    public static final Single<Object> INSTANCE = new SingleNever();
+
+    private SingleNever() {
+    }
 
     @Override
-    protected void subscribeActual(SingleObserver<? super T> s) {
+    protected void subscribeActual(SingleObserver<? super Object> s) {
         s.onSubscribe(EmptyDisposable.INSTANCE);
     }
 
