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

Author: davidmoten

src/main/java/rx/Observable.java

@@ -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());
     }
 
     /**

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,15 @@
  */
 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());
+    }
+    
+    @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,15 @@
  */
 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());
+    }
+    
+    @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,16 @@ 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>();
+    }
+    
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorSingle<T> create() {
+        return (OperatorSingle<T>) Holder.INSTANCE;
+    }
+    
+    private OperatorSingle() {
         this(false, null);
     }