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

for lift all the time.

Author: akarnokd

src/main/java/rx/Observable.java

@@ -758,7 +758,7 @@ public static final <T, R> Observable<R> combineLatest(List<? extends Observable
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava wiki: 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());
     }
 
     /**
@@ -1619,7 +1619,7 @@ public final static <T> Observable<T> merge(Iterable<? extends Observable<? exte
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Combining-Observables#merge">RxJava wiki: 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));
     }
 
     /**
@@ -1944,7 +1944,7 @@ public final static <T> Observable<T> merge(Observable<? extends T>[] sequences)
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Combining-Observables#mergedelayerror">RxJava wiki: mergeDelayError</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));
     }
 
     /**
@@ -3002,7 +3002,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());
     }
 
     /**
@@ -3923,9 +3923,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="https://github.com/ReactiveX/RxJava/wiki/Observable-Utility-Operators#dematerialize">RxJava wiki: dematerialize</a>
      */
-    @SuppressWarnings({"unchecked", "rawtypes"})
+    @SuppressWarnings({"unchecked"})
     public final <T2> Observable<T2> dematerialize() {
-        return lift(new OperatorDematerialize());
+        return lift(OperatorDematerialize.instance());
     }
 
     /**
@@ -4968,7 +4968,7 @@ private final <R> Observable<R> mapNotification(Func1<? super T, ? extends R> on
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Observable-Utility-Operators#materialize">RxJava wiki: materialize</a>
      */
     public final Observable<Notification<T>> materialize() {
-        return lift(new OperatorMaterialize<T>());
+        return lift(OperatorMaterialize.<T>instance());
     }
 
     /**

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

@@ -25,7 +25,18 @@
  *            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 an singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorAsObservable<T> instance() {
+        return (OperatorAsObservable<T>)Holder.INSTANCE;
+    }
     @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 an singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorConcat<T> instance() {
+        return (OperatorConcat<T>)Holder.INSTANCE;
+    }
+    
     @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,18 @@
  * @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 an 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
+    }
     @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,18 @@
  * 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 an singleton instance of this stateless operator.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorMaterialize<T> instance() {
+        return (OperatorMaterialize<T>)Holder.INSTANCE;
+    }
     @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 an 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;
     }