Switch (almost) all chain operators to FlowableWithUpstream. (#4318)

Author: JakeWharton

src/main/java/io/reactivex/internal/operators/flowable/FlowableAll.java

@@ -14,19 +14,16 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableAll<T> extends Flowable<Boolean> {
-    
-    final Publisher<T> source;
+public final class FlowableAll<T> extends FlowableWithUpstream<T, Boolean> {
 
     final Predicate<? super T> predicate;
 
     public FlowableAll(Publisher<T> source, Predicate<? super T> predicate) {
-        this.source = source;
+        super(source);
         this.predicate = predicate;
     }
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableAny.java

@@ -14,15 +14,13 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableAny<T> extends Flowable<Boolean> {
-    final Publisher<T> source;
+public final class FlowableAny<T> extends FlowableWithUpstream<T, Boolean> {
     final Predicate<? super T> predicate;
     public FlowableAny(Publisher<T> source, Predicate<? super T> predicate) {
-        this.source = source;
+        super(source);
         this.predicate = predicate;
     }
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundary.java

@@ -19,7 +19,6 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.disposables.*;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.fuseable.SimpleQueue;
@@ -31,15 +30,14 @@
 import io.reactivex.subscribers.*;
 
 public final class FlowableBufferBoundary<T, U extends Collection<? super T>, Open, Close> 
-extends Flowable<U> {
-    final Publisher<T> source;
+extends FlowableWithUpstream<T, U> {
     final Callable<U> bufferSupplier;
     final Publisher<? extends Open> bufferOpen;
     final Function<? super Open, ? extends Publisher<? extends Close>> bufferClose;
 
     public FlowableBufferBoundary(Publisher<T> source, Publisher<? extends Open> bufferOpen,
             Function<? super Open, ? extends Publisher<? 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/flowable/FlowableBufferBoundarySupplier.java

@@ -19,7 +19,6 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.queue.MpscLinkedQueue;
@@ -30,13 +29,12 @@
 import io.reactivex.subscribers.*;
 
 public final class FlowableBufferBoundarySupplier<T, U extends Collection<? super T>, B> 
-extends Flowable<U> {
-    final Publisher<T> source;
+extends FlowableWithUpstream<T, U> {
     final Callable<? extends Publisher<B>> boundarySupplier;
     final Callable<U> bufferSupplier;
 
     public FlowableBufferBoundarySupplier(Publisher<T> source, Callable<? extends Publisher<B>> boundarySupplier, Callable<U> bufferSupplier) {
-        this.source = source;
+        super(source);
         this.boundarySupplier = boundarySupplier;
         this.bufferSupplier = bufferSupplier;
     }

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferExactBoundary.java

@@ -18,7 +18,6 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.flowable.QueueDrainSubscriber;
@@ -27,13 +26,12 @@
 import io.reactivex.subscribers.*;
 
 public final class FlowableBufferExactBoundary<T, U extends Collection<? super T>, B> 
-extends Flowable<U> {
-    final Publisher<T> source;
+extends FlowableWithUpstream<T, U> {
     final Publisher<B> boundary;
     final Callable<U> bufferSupplier;
 
     public FlowableBufferExactBoundary(Publisher<T> source, Publisher<B> boundary, Callable<U> bufferSupplier) {
-        this.source = source;
+        super(source);
         this.boundary = boundary;
         this.bufferSupplier = bufferSupplier;
     }

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferTimed.java

@@ -29,9 +29,8 @@
 import io.reactivex.internal.util.QueueDrainHelper;
 import io.reactivex.subscribers.SerializedSubscriber;
 
-public final class FlowableBufferTimed<T, U extends Collection<? super T>> extends Flowable<U> {
+public final class FlowableBufferTimed<T, U extends Collection<? super T>> extends FlowableWithUpstream<T, U> {
 
-    final Publisher<T> source;
     final long timespan;
     final long timeskip;
     final TimeUnit unit;
@@ -42,7 +41,7 @@ public final class FlowableBufferTimed<T, U extends Collection<? super T>> exten
 
     public FlowableBufferTimed(Publisher<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/flowable/FlowableCache.java

@@ -28,7 +28,7 @@
  *
  * @param <T> the source element type
  */
-public final class FlowableCache<T> extends Flowable<T> {
+public final class FlowableCache<T> extends FlowableWithUpstream<T, T> {
     /** The cache and replay state. */
     final CacheState<T> state;
 
@@ -39,7 +39,7 @@ public final class FlowableCache<T> extends Flowable<T> {
      * @param source the source Observable to cache
      * @return the CachedObservable instance
      */
-    public static <T> FlowableCache<T> from(Flowable<? extends T> source) {
+    public static <T> FlowableCache<T> from(Flowable<T> source) {
         return from(source, 16);
     }
 
@@ -50,20 +50,21 @@ public static <T> FlowableCache<T> from(Flowable<? extends T> source) {
      * @param capacityHint the hint for the internal buffer size
      * @return the CachedObservable instance
      */
-    public static <T> FlowableCache<T> from(Flowable<? extends T> source, int capacityHint) {
+    public static <T> FlowableCache<T> from(Flowable<T> source, int capacityHint) {
         if (capacityHint < 1) {
             throw new IllegalArgumentException("capacityHint > 0 required");
         }
         CacheState<T> state = new CacheState<T>(source, capacityHint);
-        return new FlowableCache<T>(state);
+        return new FlowableCache<T>(source, state);
     }
 
     /**
      * Private constructor because state needs to be shared between the Observable body and
      * the onSubscribe function.
      * @param state
      */
-    private FlowableCache(CacheState<T> state) {
+    private FlowableCache(Flowable<T> source, CacheState<T> state) {
+        super(source);
         this.state = state;
         this.once = new AtomicBoolean();
     }

src/main/java/io/reactivex/internal/operators/flowable/FlowableCollect.java

@@ -16,19 +16,17 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.BiConsumer;
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableCollect<T, U> extends Flowable<U> {
+public final class FlowableCollect<T, U> extends FlowableWithUpstream<T, U> {
 
-    final Publisher<T> source;
     final Callable<? extends U> initialSupplier;
     final BiConsumer<? super U, ? super T> collector;
 
     public FlowableCollect(Publisher<T> source, Callable<? extends U> initialSupplier, BiConsumer<? super U, ? super T> collector) {
-        this.source = source;
+        super(source);
         this.initialSupplier = initialSupplier;
         this.collector = collector;
     }

src/main/java/io/reactivex/internal/operators/flowable/FlowableCount.java

@@ -15,15 +15,12 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableCount<T> extends Flowable<Long> {
-
-    final Publisher<T> source;
+public final class FlowableCount<T> extends FlowableWithUpstream<T, Long> {
 
     public FlowableCount(Publisher<T> source) {
-        this.source = source;
+        super(source);
     }
 
     @Override

src/main/java/io/reactivex/internal/operators/flowable/FlowableDebounce.java

@@ -17,7 +17,6 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.DisposableHelper;
@@ -26,12 +25,11 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subscribers.*;
 
-public final class FlowableDebounce<T, U> extends Flowable<T> {
-    final Publisher<T> source;
+public final class FlowableDebounce<T, U> extends FlowableWithUpstream<T, T> {
     final Function<? super T, ? extends Publisher<U>> debounceSelector;
 
     public FlowableDebounce(Publisher<T> source, Function<? super T, ? extends Publisher<U>> debounceSelector) {
-        this.source = source;
+        super(source);
         this.debounceSelector = debounceSelector;
     }