Switch (almost) all chain operators to ObservableWithUpstream. (#4319)

Author: JakeWharton

src/main/java/io/reactivex/internal/operators/observable/ObservableAll.java

@@ -18,12 +18,10 @@
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class ObservableAll<T> extends Observable<Boolean> {
-    final ObservableSource<T> source;
-    
+public final class ObservableAll<T> extends ObservableWithUpstream<T, Boolean> {
     final Predicate<? super T> predicate;
     public ObservableAll(ObservableSource<T> source, Predicate<? super T> predicate) {
-        this.source = source;
+        super(source);
         this.predicate = predicate;
     }
 

src/main/java/io/reactivex/internal/operators/observable/ObservableAny.java

@@ -17,11 +17,10 @@
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.disposables.DisposableHelper;
 
-public final class ObservableAny<T> extends Observable<Boolean> {
-    final ObservableSource<T> source;
+public final class ObservableAny<T> extends ObservableWithUpstream<T, Boolean> {
     final Predicate<? super T> predicate;
     public ObservableAny(ObservableSource<T> source, Predicate<? super T> predicate) {
-        this.source = source;
+        super(source);
         this.predicate = predicate;
     }
 

src/main/java/io/reactivex/internal/operators/observable/ObservableBuffer.java

@@ -17,20 +17,18 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import io.reactivex.Observable;
 import io.reactivex.ObservableSource;
 import io.reactivex.Observer;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.*;
 
-public final class ObservableBuffer<T, U extends Collection<? super T>> extends Observable<U> {
-    final ObservableSource<T> source;
+public final class ObservableBuffer<T, U extends Collection<? super T>> extends ObservableWithUpstream<T, U> {
     final int count;
     final int skip;
     final Callable<U> bufferSupplier;
 
     public ObservableBuffer(ObservableSource<T> source, int count, int skip, Callable<U> bufferSupplier) {
-        this.source = source;
+        super(source);
         this.count = count;
         this.skip = skip;
         this.bufferSupplier = bufferSupplier;

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundary.java

@@ -17,7 +17,6 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import io.reactivex.Observable;
 import io.reactivex.ObservableSource;
 import io.reactivex.Observer;
 import io.reactivex.disposables.*;
@@ -31,15 +30,14 @@
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableBufferBoundary<T, U extends Collection<? super T>, Open, Close> 
-extends Observable<U> {
-    final ObservableSource<T> source;
+extends ObservableWithUpstream<T, U> {
     final Callable<U> bufferSupplier;
     final ObservableSource<? extends Open> bufferOpen;
     final Function<? super Open, ? extends ObservableSource<? extends Close>> bufferClose;
 
     public ObservableBufferBoundary(ObservableSource<T> source, ObservableSource<? extends Open> bufferOpen,
                                     Function<? super Open, ? extends ObservableSource<? extends Close>> bufferClose, Callable<U> bufferSupplier) {
-        this.source = source;
+        super(source);
         this.bufferOpen = bufferOpen;
         this.bufferClose = bufferClose;
         this.bufferSupplier = bufferSupplier;

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundarySupplier.java

@@ -27,13 +27,12 @@
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableBufferBoundarySupplier<T, U extends Collection<? super T>, B> 
-extends Observable<U> {
-    final ObservableSource<T> source;
+extends ObservableWithUpstream<T, U> {
     final Callable<? extends ObservableSource<B>> boundarySupplier;
     final Callable<U> bufferSupplier;
 
     public ObservableBufferBoundarySupplier(ObservableSource<T> source, Callable<? extends ObservableSource<B>> boundarySupplier, Callable<U> bufferSupplier) {
-        this.source = source;
+        super(source);
         this.boundarySupplier = boundarySupplier;
         this.bufferSupplier = bufferSupplier;
     }

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferExactBoundary.java

@@ -25,13 +25,12 @@
 import io.reactivex.observers.SerializedObserver;
 
 public final class ObservableBufferExactBoundary<T, U extends Collection<? super T>, B> 
-extends Observable<U> {
-    final ObservableSource<T> source;
+extends ObservableWithUpstream<T, U> {
     final ObservableSource<B> boundary;
     final Callable<U> bufferSupplier;
 
     public ObservableBufferExactBoundary(ObservableSource<T> source, ObservableSource<B> boundary, Callable<U> bufferSupplier) {
-        this.source = source;
+        super(source);
         this.boundary = boundary;
         this.bufferSupplier = bufferSupplier;
     }

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferTimed.java

@@ -18,7 +18,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.*;
-import io.reactivex.Observable;
 import io.reactivex.Observer;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
@@ -29,9 +28,8 @@
 import io.reactivex.observers.SerializedObserver;
 
 public final class ObservableBufferTimed<T, U extends Collection<? super T>> 
-extends Observable<U> {
+extends ObservableWithUpstream<T, U> {
 
-    final ObservableSource<T> source;
     final long timespan;
     final long timeskip;
     final TimeUnit unit;
@@ -42,7 +40,7 @@ public final class ObservableBufferTimed<T, U extends Collection<? super T>>
 
     public ObservableBufferTimed(ObservableSource<T> source, long timespan, long timeskip, TimeUnit unit, Scheduler scheduler, Callable<U> bufferSupplier, int maxSize,
                                  boolean restartTimerOnMaxSize) {
-        this.source = source;
+        super(source);
         this.timespan = timespan;
         this.timeskip = timeskip;
         this.unit = unit;

src/main/java/io/reactivex/internal/operators/observable/ObservableCache.java

@@ -25,7 +25,7 @@
  *
  * @param <T> the source element type
  */
-public final class ObservableCache<T> extends Observable<T> {
+public final class ObservableCache<T> extends ObservableWithUpstream<T, T> {
     /** The cache and replay state. */
     final CacheState<T> state;
 
@@ -37,7 +37,7 @@ public final class ObservableCache<T> extends Observable<T> {
      * @param source the source Observable to cache
      * @return the CachedObservable instance
      */
-    public static <T> ObservableCache<T> from(Observable<? extends T> source) {
+    public static <T> ObservableCache<T> from(Observable<T> source) {
         return from(source, 16);
     }
 
@@ -48,12 +48,12 @@ public static <T> ObservableCache<T> from(Observable<? extends T> source) {
      * @param capacityHint the hint for the internal buffer size
      * @return the CachedObservable instance
      */
-    public static <T> ObservableCache<T> from(Observable<? extends T> source, int capacityHint) {
+    public static <T> ObservableCache<T> from(Observable<T> source, int capacityHint) {
         if (capacityHint < 1) {
             throw new IllegalArgumentException("capacityHint > 0 required");
         }
         CacheState<T> state = new CacheState<T>(source, capacityHint);
-        return new ObservableCache<T>(state);
+        return new ObservableCache<T>(source, state);
     }
 
     /**
@@ -62,7 +62,8 @@ public static <T> ObservableCache<T> from(Observable<? extends T> source, int ca
      * @param onSubscribe
      * @param state
      */
-    private ObservableCache(CacheState<T> state) {
+    private ObservableCache(Observable<T> source, CacheState<T> state) {
+        super(source);
         this.state = state;
         this.once = new AtomicBoolean();
     }

src/main/java/io/reactivex/internal/operators/observable/ObservableLift.java

@@ -25,14 +25,12 @@
  * @param <T> the upstream value type
  * @param <R> the downstream parameter type
  */
-public final class ObservableLift<R, T> extends Observable<R> {
+public final class ObservableLift<R, T> extends ObservableWithUpstream<T, R> {
     /** The actual operator. */
     final ObservableOperator<? extends R, ? super T> operator;
-    /** The source publisher. */
-    final ObservableSource<? extends T> source;
-    
-    public ObservableLift(ObservableSource<? extends T> source, ObservableOperator<? extends R, ? super T> operator) {
-        this.source = source;
+
+    public ObservableLift(ObservableSource<T> source, ObservableOperator<? extends R, ? super T> operator) {
+        super(source);
         this.operator = operator;
     }
 
@@ -44,14 +42,6 @@ public ObservableLift(ObservableSource<? extends T> source, ObservableOperator<?
         return operator;
     }
 
-    /**
-     * Returns the source of this lift publisher.
-     * @return the source of this lift publisher
-     */
-    public ObservableSource<? extends T> source() {
-        return source;
-    }
-    
     @Override
     public void subscribeActual(Observer<? super R> s) {
         try {

src/main/java/io/reactivex/internal/operators/observable/ObservableRedo.java

@@ -22,13 +22,12 @@
 import io.reactivex.internal.subscribers.observable.ToNotificationObserver;
 import io.reactivex.subjects.BehaviorSubject;
 
-public final class ObservableRedo<T> extends Observable<T> {
-    final ObservableSource<? extends T> source;
+public final class ObservableRedo<T> extends ObservableWithUpstream<T, T> {
     final Function<? super Observable<Try<Optional<Object>>>, ? extends ObservableSource<?>> manager;
 
-    public ObservableRedo(ObservableSource<? extends T> source,
+    public ObservableRedo(ObservableSource<T> source,
             Function<? super Observable<Try<Optional<Object>>>, ? extends ObservableSource<?>> manager) {
-        this.source = source;
+        super(source);
         this.manager = manager;
     }