Merge pull request #2335 from akarnokd/StatelessOperators

A set of stateless operators that don't need to be instantiated

Author: akarnokd

src/main/java/rx/Observable.java

@@ -759,7 +759,7 @@ public static final <T, R> Observable<R> combineLatest(List<? extends Observable
      * @see <a href="http://reactivex.io/documentation/operators/concat.html">ReactiveX operators documentation: Concat</a>
      */
     public final static <T> Observable<T> concat(Observable<? extends Observable<? extends T>> observables) {
-        return observables.lift(new OperatorConcat<T>());
+        return observables.lift(OperatorConcat.<T>instance());
     }
 
     /**
@@ -1620,7 +1620,7 @@ public final static <T> Observable<T> merge(Iterable<? extends Observable<? exte
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
-        return source.lift(new OperatorMerge<T>());
+        return source.lift(OperatorMerge.<T>instance(false));
     }
 
     /**
@@ -1945,7 +1945,7 @@ public final static <T> Observable<T> merge(Observable<? extends T>[] sequences)
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
     public final static <T> Observable<T> mergeDelayError(Observable<? extends Observable<? extends T>> source) {
-        return source.lift(new OperatorMergeDelayError<T>());
+        return source.lift(OperatorMerge.<T>instance(true));
     }
 
     /**
@@ -2421,7 +2421,7 @@ public final static <T> Observable<Boolean> sequenceEqual(Observable<? extends T
      * @see <a href="http://reactivex.io/documentation/operators/switch.html">ReactiveX operators documentation: Switch</a>
      */
     public final static <T> Observable<T> switchOnNext(Observable<? extends Observable<? extends T>> sequenceOfSequences) {
-        return sequenceOfSequences.lift(new OperatorSwitch<T>());
+        return sequenceOfSequences.lift(OperatorSwitch.<T>instance());
     }
 
     /**
@@ -3004,7 +3004,7 @@ public final Observable<T> ambWith(Observable<? extends T> t1) {
      * @return an Observable that hides the identity of this Observable
      */
     public final Observable<T> asObservable() {
-        return lift(new OperatorAsObservable<T>());
+        return lift(OperatorAsObservable.<T>instance());
     }
 
     /**
@@ -3925,9 +3925,9 @@ public final <U> Observable<T> delaySubscription(Func0<? extends Observable<U>>
      *             if the source Observable is not of type {@code Observable<Notification<T>>}
      * @see <a href="http://reactivex.io/documentation/operators/materialize-dematerialize.html">ReactiveX operators documentation: Dematerialize</a>
      */
-    @SuppressWarnings({"unchecked", "rawtypes"})
+    @SuppressWarnings({"unchecked"})
     public final <T2> Observable<T2> dematerialize() {
-        return lift(new OperatorDematerialize());
+        return lift(OperatorDematerialize.instance());
     }
 
     /**
@@ -4963,7 +4963,7 @@ private final <R> Observable<R> mapNotification(Func1<? super T, ? extends R> on
      * @see <a href="http://reactivex.io/documentation/operators/materialize-dematerialize.html">ReactiveX operators documentation: Materialize</a>
      */
     public final Observable<Notification<T>> materialize() {
-        return lift(new OperatorMaterialize<T>());
+        return lift(OperatorMaterialize.<T>instance());
     }
 
     /**
@@ -5108,7 +5108,7 @@ public final Observable<T> onBackpressureBuffer(long capacity, Action0 onOverflo
      * @see <a href="http://reactivex.io/documentation/operators/backpressure.html">ReactiveX operators documentation: backpressure operators</a>
      */
     public final Observable<T> onBackpressureDrop() {
-        return lift(new OperatorOnBackpressureDrop<T>());
+        return lift(OperatorOnBackpressureDrop.<T>instance());
     }
 
     /**
@@ -6458,7 +6458,7 @@ public final <R> Observable<R> scan(R initialValue, Func2<R, ? super T, R> accum
      * @see <a href="http://reactivex.io/documentation/operators/serialize.html">ReactiveX operators documentation: Serialize</a>
      */
     public final Observable<T> serialize() {
-        return lift(new OperatorSerialize<T>());
+        return lift(OperatorSerialize.<T>instance());
     }
 
     /**
@@ -8362,7 +8362,7 @@ public final BlockingObservable<T> toBlocking() {
      * @see <a href="http://reactivex.io/documentation/operators/to.html">ReactiveX operators documentation: To</a>
      */
     public final Observable<List<T>> toList() {
-        return lift(new OperatorToObservableList<T>());
+        return lift(OperatorToObservableList.<T>instance());
     }
 
     /**

src/main/java/rx/internal/operators/OperatorAsObservable.java

@@ -25,7 +25,19 @@
  *            the return value type of the wrapped observable.
  */
 public final class OperatorAsObservable<T> implements Operator<T, T> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorAsObservable<Object> INSTANCE = new OperatorAsObservable<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorAsObservable<T> instance() {
+        return (OperatorAsObservable<T>)Holder.INSTANCE;
+    }
+    private OperatorAsObservable() { }
     @Override
     public Subscriber<? super T> call(Subscriber<? super T> s) {
         return s;

src/main/java/rx/internal/operators/OperatorConcat.java

@@ -37,6 +37,19 @@
  *            the source and result value type
  */
 public final class OperatorConcat<T> implements Operator<T, Observable<? extends T>> {
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorConcat<Object> INSTANCE = new OperatorConcat<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorConcat<T> instance() {
+        return (OperatorConcat<T>)Holder.INSTANCE;
+    }
+    private OperatorConcat() { }
     @Override
     public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
         final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);

src/main/java/rx/internal/operators/OperatorDematerialize.java

@@ -30,7 +30,19 @@
  * @param <T> the wrapped value type
  */
 public final class OperatorDematerialize<T> implements Operator<T, Notification<T>> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorDematerialize<Object> INSTANCE = new OperatorDematerialize<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    public static OperatorDematerialize instance() {
+        return Holder.INSTANCE; // using raw types because the type inference is not good enough
+    }
+    private OperatorDematerialize() { }
     @Override
     public Subscriber<? super Notification<T>> call(final Subscriber<? super T> child) {
         return new Subscriber<Notification<T>>(child) {

src/main/java/rx/internal/operators/OperatorMaterialize.java

@@ -29,7 +29,19 @@
  * See <a href="http://msdn.microsoft.com/en-us/library/hh229453.aspx">here</a> for the Microsoft Rx equivalent.
  */
 public final class OperatorMaterialize<T> implements Operator<Notification<T>, T> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorMaterialize<Object> INSTANCE = new OperatorMaterialize<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorMaterialize<T> instance() {
+        return (OperatorMaterialize<T>)Holder.INSTANCE;
+    }
+    private OperatorMaterialize() { }
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Notification<T>> child) {
         return new Subscriber<T>(child) {

src/main/java/rx/internal/operators/OperatorMerge.java

@@ -38,12 +38,46 @@
  * <img width="640" height="380" src="https://raw.githubusercontent.com/wiki/ReactiveX/RxJava/images/rx-operators/merge.png" alt="">
  * <p>
  * You can combine the items emitted by multiple {@code Observable}s so that they act like a single {@code Observable}, by using the merge operation.
- * 
+ * <p>
+ * The {@code instance(true)} call behaves like {@link OperatorMerge} except that if any of the merged Observables notify of
+ * an error via {@code onError}, {@code mergeDelayError} will refrain from propagating that error
+ * notification until all of the merged Observables have finished emitting items.
+ * <p>
+ * <img width="640" src="https://github.com/ReactiveX/RxJava/wiki/images/rx-operators/mergeDelayError.png" alt="">
+ * <p>
+ * Even if multiple merged Observables send {@code onError} notifications, {@code mergeDelayError} will
+ * only invoke the {@code onError} method of its Observers once.
+ * <p>
+ * This operation allows an Observer to receive all successfully emitted items from all of the
+ * source Observables without being interrupted by an error notification from one of them.
+ * <p>
+ * <em>Note:</em> If this is used on an Observable that never completes, it will never call {@code onError} and will effectively swallow errors.
+
  * @param <T>
  *            the type of the items emitted by both the source and merged {@code Observable}s
  */
 public class OperatorMerge<T> implements Operator<T, Observable<? extends T>> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class HolderNoDelay {
+        /** A singleton instance. */
+        static final OperatorMerge<Object> INSTANCE = new OperatorMerge<Object>(false);
+    }
+    /** Lazy initialization via inner-class holder. */
+    private static final class HolderDelayErrors {
+        /** A singleton instance. */
+        static final OperatorMerge<Object> INSTANCE = new OperatorMerge<Object>(true);
+    }
+    /**
+     * @param delayErrors should the merge delay errors?
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorMerge<T> instance(boolean delayErrors) {
+        if (delayErrors) {
+            return (OperatorMerge<T>)HolderDelayErrors.INSTANCE;
+        }
+        return (OperatorMerge<T>)HolderNoDelay.INSTANCE;
+    }
     /*
      * benjchristensen => This class is complex and I'm not a fan of it despite writing it. I want to give some background
      * as to why for anyone who wants to try and help improve it.
@@ -74,11 +108,11 @@ public class OperatorMerge<T> implements Operator<T, Observable<? extends T>> {
      * to track object allocation.
      */
 
-    public OperatorMerge() {
+    private OperatorMerge() {
         this.delayErrors = false;
     }
 
-    public OperatorMerge(boolean delayErrors) {
+    private OperatorMerge(boolean delayErrors) {
         this.delayErrors = delayErrors;
     }
 

src/main/java/rx/internal/operators/OperatorMergeDelayError.java

@@ -22,7 +22,19 @@
 import rx.Subscriber;
 
 public class OperatorOnBackpressureDrop<T> implements Operator<T, T> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorOnBackpressureDrop<Object> INSTANCE = new OperatorOnBackpressureDrop<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings({ "unchecked" })
+    public static <T> OperatorOnBackpressureDrop<T> instance() {
+        return (OperatorOnBackpressureDrop<T>)Holder.INSTANCE;
+    }
+    private OperatorOnBackpressureDrop() { }
     @Override
     public Subscriber<? super T> call(final Subscriber<? super T> child) {
         final AtomicLong requested = new AtomicLong();

src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java

@@ -20,7 +20,19 @@
 import rx.observers.SerializedSubscriber;
 
 public final class OperatorSerialize<T> implements Operator<T, T> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorSerialize<Object> INSTANCE = new OperatorSerialize<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings({ "unchecked" })
+    public static <T> OperatorSerialize<T> instance() {
+        return (OperatorSerialize<T>)Holder.INSTANCE;
+    }
+    private OperatorSerialize() { }
     @Override
     public Subscriber<? super T> call(final Subscriber<? super T> s) {
         return new SerializedSubscriber<T>(new Subscriber<T>(s) {

src/main/java/rx/internal/operators/OperatorSerialize.java

@@ -17,6 +17,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+
 import rx.Observable;
 import rx.Observable.Operator;
 import rx.Producer;
@@ -33,7 +34,19 @@
  * @param <T> the value type
  */
 public final class OperatorSwitch<T> implements Operator<T, Observable<? extends T>> {
-
+    /** Lazy initialization via inner-class holder. */
+    private static final class Holder {
+        /** A singleton instance. */
+        static final OperatorSwitch<Object> INSTANCE = new OperatorSwitch<Object>();
+    }
+    /**
+     * @return a singleton instance of this stateless operator.
+     */
+    @SuppressWarnings({ "unchecked" })
+    public static <T> OperatorSwitch<T> instance() {
+        return (OperatorSwitch<T>)Holder.INSTANCE;
+    }
+    private OperatorSwitch() { }
     @Override
     public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
         return new SwitchSubscriber<T>(child);