reduce gc pressure by using singleton Operators for single, distinct, distinctUntilChanged, onBackpressureBuffer, isEmpty

Author: davidmoten

src/main/java/rx/Observable.java

@@ -2295,7 +2295,7 @@ public final Observable<Observable<T>> nest() {
      * @see <a href="http://reactivex.io/documentation/operators/empty-never-throw.html">ReactiveX operators documentation: Never</a>
      */
     public final static <T> Observable<T> never() {
-        return new NeverObservable<T>();
+        return NeverObservable.create();
     }
 
     /**
@@ -4026,7 +4026,7 @@ public final <T2> Observable<T2> dematerialize() {
      * @see <a href="http://reactivex.io/documentation/operators/distinct.html">ReactiveX operators documentation: Distinct</a>
      */
     public final Observable<T> distinct() {
-        return lift(new OperatorDistinct<T, T>(UtilityFunctions.<T>identity()));
+        return lift(OperatorDistinct.<T> create());
     }
 
     /**
@@ -4064,7 +4064,7 @@ public final <U> Observable<T> distinct(Func1<? super T, ? extends U> keySelecto
      * @see <a href="http://reactivex.io/documentation/operators/distinct.html">ReactiveX operators documentation: Distinct</a>
      */
     public final Observable<T> distinctUntilChanged() {
-        return lift(new OperatorDistinctUntilChanged<T, T>(UtilityFunctions.<T>identity()));
+        return lift(OperatorDistinctUntilChanged.<T> create());
     }
 
     /**
@@ -4959,8 +4959,13 @@ public final Observable<T> ignoreElements() {
      * @return an Observable that emits a Boolean
      * @see <a href="http://reactivex.io/documentation/operators/contains.html">ReactiveX operators documentation: Contains</a>
      */
+    @SuppressWarnings("unchecked")
     public final Observable<Boolean> isEmpty() {
-        return lift(new OperatorAny<T>(UtilityFunctions.alwaysTrue(), true));
+        return lift((OperatorAny<T>) HolderAnyForEmpty.INSTANCE);
+    }
+    
+    private static class HolderAnyForEmpty {
+        static final OperatorAny<?> INSTANCE = new OperatorAny<Object>(UtilityFunctions.alwaysTrue(), true);
     }
 
     /**
@@ -5226,7 +5231,7 @@ public final Boolean call(T t) {
      * @see <a href="http://reactivex.io/documentation/operators/backpressure.html">ReactiveX operators documentation: backpressure operators</a>
      */
     public final Observable<T> onBackpressureBuffer() {
-        return lift(new OperatorOnBackpressureBuffer<T>());
+        return lift(OperatorOnBackpressureBuffer.<T> create());
     }
 
     /**
@@ -6709,7 +6714,7 @@ public final Observable<T> share() {
      * @see <a href="http://reactivex.io/documentation/operators/first.html">ReactiveX operators documentation: First</a>
      */
     public final Observable<T> single() {
-        return lift(new OperatorSingle<T>());
+        return lift(OperatorSingle.<T>create());
     }
 
     /**
@@ -9276,7 +9281,22 @@ public final <T2, R> Observable<R> zipWith(Observable<? extends T2> other, Func2
      *            the type of item (not) emitted by the Observable
      */
     private static class NeverObservable<T> extends Observable<T> {
-        public NeverObservable() {
+        
+        private static class Holder {
+            static final NeverObservable<?> INSTANCE = new NeverObservable<Object>();
+        }
+        
+        /**
+         * Returns a singleton instance of NeverObservble (cast to the generic type).
+         * 
+         * @return
+         */
+        @SuppressWarnings("unchecked")
+        static <T> NeverObservable<T> create() {
+            return (NeverObservable<T>) Holder.INSTANCE;
+        }
+        
+        NeverObservable() {
             super(new OnSubscribe<T>() {
 
                 @Override

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

@@ -17,9 +17,11 @@
 
 import java.util.HashSet;
 import java.util.Set;
+
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.functions.Func1;
+import rx.internal.util.UtilityFunctions;
 
 /**
  * Returns an Observable that emits all distinct items emitted by the source.
@@ -29,6 +31,21 @@
  */
 public final class OperatorDistinct<T, U> implements Operator<T, T> {
     final Func1<? super T, ? extends U> keySelector;
+    
+    private static class Holder {
+        static final OperatorDistinct<?,?> INSTANCE = new OperatorDistinct<Object,Object>(UtilityFunctions.<Object>identity());
+    }
+    
+    /**
+     * Returns a singleton instance of OperatorDistinct that was built using 
+     * the identity function for comparison (<code>new OperatorDistinct(UtilityFunctions.identity())</code>).
+     * 
+     * @return Operator that emits distinct values only (regardless of order) using the identity function for comparison 
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorDistinct<T,T> create() {
+        return (OperatorDistinct<T, T>) Holder.INSTANCE;
+    }
 
     public OperatorDistinct(Func1<? super T, ? extends U> keySelector) {
         this.keySelector = keySelector;

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

@@ -18,6 +18,7 @@
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.functions.Func1;
+import rx.internal.util.UtilityFunctions;
 
 /**
  * Returns an Observable that emits all sequentially distinct items emitted by the source.
@@ -26,6 +27,22 @@
  */
 public final class OperatorDistinctUntilChanged<T, U> implements Operator<T, T> {
     final Func1<? super T, ? extends U> keySelector;
+    
+    private static class Holder {
+        static final OperatorDistinctUntilChanged<?,?> INSTANCE = new OperatorDistinctUntilChanged<Object,Object>(UtilityFunctions.identity());
+    }
+    
+
+    /**
+     * Returns a singleton instance of OperatorDistinctUntilChanged that was built using 
+     * the identity function for comparison (<code>new OperatorDistinctUntilChanged(UtilityFunctions.identity())</code>).
+     * 
+     * @return Operator that emits sequentially distinct values only using the identity function for comparison 
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorDistinctUntilChanged<T,T> create() {
+        return (OperatorDistinctUntilChanged<T, T>) Holder.INSTANCE;
+    }
 
     public OperatorDistinctUntilChanged(Func1<? super T, ? extends U> keySelector) {
         this.keySelector = keySelector;

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

@@ -31,7 +31,16 @@ public class OperatorOnBackpressureBuffer<T> implements Operator<T, T> {
     private final Long capacity;
     private final Action0 onOverflow;
 
-    public OperatorOnBackpressureBuffer() {
+    private static class Holder {
+        static final OperatorOnBackpressureBuffer<?> INSTANCE = new OperatorOnBackpressureBuffer<Object>();
+    }
+    
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorOnBackpressureBuffer<T> create() {
+        return (OperatorOnBackpressureBuffer<T>) Holder.INSTANCE;
+    }
+    
+    private OperatorOnBackpressureBuffer() {
         this.capacity = null;
         this.onOverflow = null;
     }

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

@@ -32,7 +32,22 @@ public final class OperatorSingle<T> implements Operator<T, T> {
     private final boolean hasDefaultValue;
     private final T defaultValue;
 
-    public OperatorSingle() {
+    private static class Holder {
+        final static OperatorSingle<?> INSTANCE = new OperatorSingle<Object>();
+    }
+    
+    /**
+     * Returns a singleton instance of OperatorSingle (if the stream is empty or has 
+     * more than one element an error will be emitted) that is cast to the generic type.
+     *  
+     * @return a singleton instance of an Operator that will emit a single value only unless the stream has zero or more than one element in which case it will emit an error.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorSingle<T> create() {
+        return (OperatorSingle<T>) Holder.INSTANCE;
+    }
+    
+    private OperatorSingle() {
         this(false, null);
     }