Extract 'WithUpstream' interfaces. (#4326)

This allows use with types that do not extend directly from their base stream types.

Author: JakeWharton

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

@@ -0,0 +1,48 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.flowable;
+
+import org.reactivestreams.Publisher;
+
+import io.reactivex.Flowable;
+import io.reactivex.internal.functions.Objects;
+
+/**
+ * Abstract base class for operators that take an upstream
+ * source {@link Publisher}.
+ *
+ * @param <T> the upstream value type
+ * @param <R> the output value type
+ */
+abstract class AbstractFlowableWithUpstream<T, R> extends Flowable<R> implements FlowableWithUpstream<T> {
+    
+    /**
+     * The upstream source Publisher.
+     */
+    protected final Publisher<T> source;
+    
+    /**
+     * Constructs a FlowableSource wrapping the given non-null (verified)
+     * source Publisher.
+     * @param source the source (upstream) Publisher instance, not null (verified)
+     */
+    public AbstractFlowableWithUpstream(Publisher<T> source) {
+        this.source = Objects.requireNonNull(source, "source is null");
+    }
+    
+    @Override
+    public final Publisher<T> source() {
+        return source;
+    }
+}

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

@@ -18,7 +18,7 @@
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableAll<T> extends FlowableWithUpstream<T, Boolean> {
+public final class FlowableAll<T> extends AbstractFlowableWithUpstream<T, Boolean> {
 
     final Predicate<? super T> predicate;
 

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

@@ -17,7 +17,7 @@
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableAny<T> extends FlowableWithUpstream<T, Boolean> {
+public final class FlowableAny<T> extends AbstractFlowableWithUpstream<T, Boolean> {
     final Predicate<? super T> predicate;
     public FlowableAny(Publisher<T> source, Predicate<? super T> predicate) {
         super(source);

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

@@ -26,7 +26,7 @@
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableBuffer<T, C extends Collection<? super T>> extends FlowableWithUpstream<T, C> {
+public final class FlowableBuffer<T, C extends Collection<? super T>> extends AbstractFlowableWithUpstream<T, C> {
     final int size;
 
     final int skip;

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

@@ -30,7 +30,7 @@
 import io.reactivex.subscribers.*;
 
 public final class FlowableBufferBoundary<T, U extends Collection<? super T>, Open, Close> 
-extends FlowableWithUpstream<T, U> {
+extends AbstractFlowableWithUpstream<T, U> {
     final Callable<U> bufferSupplier;
     final Publisher<? extends Open> bufferOpen;
     final Function<? super Open, ? extends Publisher<? extends Close>> bufferClose;

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

@@ -29,7 +29,7 @@
 import io.reactivex.subscribers.*;
 
 public final class FlowableBufferBoundarySupplier<T, U extends Collection<? super T>, B> 
-extends FlowableWithUpstream<T, U> {
+extends AbstractFlowableWithUpstream<T, U> {
     final Callable<? extends Publisher<B>> boundarySupplier;
     final Callable<U> bufferSupplier;
 

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

@@ -26,7 +26,7 @@
 import io.reactivex.subscribers.*;
 
 public final class FlowableBufferExactBoundary<T, U extends Collection<? super T>, B> 
-extends FlowableWithUpstream<T, U> {
+extends AbstractFlowableWithUpstream<T, U> {
     final Publisher<B> boundary;
     final Callable<U> bufferSupplier;
 

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

@@ -29,7 +29,7 @@
 import io.reactivex.internal.util.QueueDrainHelper;
 import io.reactivex.subscribers.SerializedSubscriber;
 
-public final class FlowableBufferTimed<T, U extends Collection<? super T>> extends FlowableWithUpstream<T, U> {
+public final class FlowableBufferTimed<T, U extends Collection<? super T>> extends AbstractFlowableWithUpstream<T, U> {
 
     final long timespan;
     final long timeskip;

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 FlowableWithUpstream<T, T> {
+public final class FlowableCache<T> extends AbstractFlowableWithUpstream<T, T> {
     /** The cache and replay state. */
     final CacheState<T> state;
 

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

@@ -20,7 +20,7 @@
 import io.reactivex.functions.BiConsumer;
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableCollect<T, U> extends FlowableWithUpstream<T, U> {
+public final class FlowableCollect<T, U> extends AbstractFlowableWithUpstream<T, U> {
 
     final Callable<? extends U> initialSupplier;
     final BiConsumer<? super U, ? super T> collector;

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

@@ -26,7 +26,7 @@
 import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableConcatMap<T, R> extends FlowableWithUpstream<T, R> {
+public final class FlowableConcatMap<T, R> extends AbstractFlowableWithUpstream<T, R> {
 
     final Function<? super T, ? extends Publisher<? extends R>> mapper;
 

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

@@ -28,7 +28,7 @@
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public class FlowableConcatMapEager<T, R> extends FlowableWithUpstream<T, R> {
+public class FlowableConcatMapEager<T, R> extends AbstractFlowableWithUpstream<T, R> {
 
     final Function<? super T, ? extends Publisher<? extends R>> mapper;
 

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

@@ -17,7 +17,7 @@
 
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableCount<T> extends FlowableWithUpstream<T, Long> {
+public final class FlowableCount<T> extends AbstractFlowableWithUpstream<T, Long> {
 
     public FlowableCount(Publisher<T> source) {
         super(source);

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

@@ -25,7 +25,7 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subscribers.*;
 
-public final class FlowableDebounce<T, U> extends FlowableWithUpstream<T, T> {
+public final class FlowableDebounce<T, U> extends AbstractFlowableWithUpstream<T, T> {
     final Function<? super T, ? extends Publisher<U>> debounceSelector;
 
     public FlowableDebounce(Publisher<T> source, Function<? super T, ? extends Publisher<U>> debounceSelector) {

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

@@ -27,7 +27,7 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subscribers.SerializedSubscriber;
 
-public final class FlowableDebounceTimed<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableDebounceTimed<T> extends AbstractFlowableWithUpstream<T, T> {
     final long timeout;
     final TimeUnit unit;
     final Scheduler scheduler;

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

@@ -22,7 +22,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.subscribers.SerializedSubscriber;
 
-public final class FlowableDelay<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableDelay<T> extends AbstractFlowableWithUpstream<T, T> {
     final long delay;
     final TimeUnit unit;
     final Scheduler scheduler;

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

@@ -19,7 +19,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableDematerialize<T> extends FlowableWithUpstream<Try<Optional<T>>, T> {
+public final class FlowableDematerialize<T> extends AbstractFlowableWithUpstream<Try<Optional<T>>, T> {
 
     public FlowableDematerialize(Publisher<Try<Optional<T>>> source) {
         super(source);

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

@@ -18,7 +18,7 @@
 import io.reactivex.internal.subscribers.flowable.EmptyComponent;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 
-public final class FlowableDetach<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableDetach<T> extends AbstractFlowableWithUpstream<T, T> {
 
     public FlowableDetach(Publisher<T> source) {
         super(source);

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

@@ -23,7 +23,7 @@
 import io.reactivex.internal.functions.*;
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableDistinct<T, K> extends FlowableWithUpstream<T, T> {
+public final class FlowableDistinct<T, K> extends AbstractFlowableWithUpstream<T, T> {
     final Function<? super T, K> keySelector;
     final Callable<? extends Predicate<? super K>> predicateSupplier;
 

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

@@ -19,7 +19,7 @@
 import io.reactivex.internal.fuseable.ConditionalSubscriber;
 import io.reactivex.internal.subscribers.flowable.*;
 
-public final class FlowableDistinctUntilChanged<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableDistinctUntilChanged<T> extends AbstractFlowableWithUpstream<T, T> {
 
     final BiPredicate<? super T, ? super T> comparer;
 

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

@@ -21,7 +21,7 @@
 import io.reactivex.internal.subscribers.flowable.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableDoOnEach<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableDoOnEach<T> extends AbstractFlowableWithUpstream<T, T> {
     final Consumer<? super T> onNext;
     final Consumer<? super Throwable> onError;
     final Runnable onComplete;

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

@@ -19,7 +19,7 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
-public final class FlowableDoOnLifecycle<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableDoOnLifecycle<T> extends AbstractFlowableWithUpstream<T, T> {
     private final Consumer<? super Subscription> onSubscribe;
     private final LongConsumer onRequest;
     private final Runnable onCancel;

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

@@ -17,7 +17,7 @@
 
 import io.reactivex.internal.subscriptions.*;
 
-public final class FlowableElementAt<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableElementAt<T> extends AbstractFlowableWithUpstream<T, T> {
     final long index;
     final T defaultValue;
     public FlowableElementAt(Publisher<T> source, long index, T defaultValue) {

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

@@ -19,7 +19,7 @@
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.subscribers.flowable.*;
 
-public final class FlowableFilter<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableFilter<T> extends AbstractFlowableWithUpstream<T, T> {
     final Predicate<? super T> predicate;
     public FlowableFilter(Publisher<T> source, Predicate<? super T> predicate) {
         super(source);

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

@@ -26,7 +26,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableFlatMap<T, U> extends FlowableWithUpstream<T, U> {
+public final class FlowableFlatMap<T, U> extends AbstractFlowableWithUpstream<T, U> {
     final Function<? super T, ? extends Publisher<? extends U>> mapper;
     final boolean delayErrors;
     final int maxConcurrency;

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

@@ -28,7 +28,7 @@
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableFlattenIterable<T, R> extends FlowableWithUpstream<T, R> {
+public final class FlowableFlattenIterable<T, R> extends AbstractFlowableWithUpstream<T, R> {
 
     final Function<? super T, ? extends Iterable<? extends R>> mapper;
 

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

@@ -28,7 +28,7 @@
 import io.reactivex.internal.util.BackpressureHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableGroupBy<T, K, V> extends FlowableWithUpstream<T, GroupedFlowable<K, V>> {
+public final class FlowableGroupBy<T, K, V> extends AbstractFlowableWithUpstream<T, GroupedFlowable<K, V>> {
     final Function<? super T, ? extends K> keySelector;
     final Function<? super T, ? extends V> valueSelector;
     final int bufferSize;

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

@@ -33,7 +33,7 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.UnicastProcessor;
 
-public class FlowableGroupJoin<TLeft, TRight, TLeftEnd, TRightEnd, R> extends FlowableWithUpstream<TLeft, R> {
+public class FlowableGroupJoin<TLeft, TRight, TLeftEnd, TRightEnd, R> extends AbstractFlowableWithUpstream<TLeft, R> {
 
     final Publisher<? extends TRight> other;
 

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

@@ -23,7 +23,7 @@
  * 
  * @since 2.0
  */
-public class FlowableHide<T> extends FlowableWithUpstream<T, T> {
+public class FlowableHide<T> extends AbstractFlowableWithUpstream<T, T> {
 
     public FlowableHide(Publisher<T> source) {
         super(source);

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

@@ -18,7 +18,7 @@
 import io.reactivex.internal.fuseable.QueueSubscription;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 
-public final class FlowableIgnoreElements<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableIgnoreElements<T> extends AbstractFlowableWithUpstream<T, T> {
 
     public FlowableIgnoreElements(Publisher<T> source) {
         super(source);

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

@@ -29,7 +29,7 @@
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public class FlowableJoin<TLeft, TRight, TLeftEnd, TRightEnd, R> extends FlowableWithUpstream<TLeft, R> {
+public class FlowableJoin<TLeft, TRight, TLeftEnd, TRightEnd, R> extends AbstractFlowableWithUpstream<TLeft, R> {
 
     final Publisher<? extends TRight> other;
 

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

@@ -27,7 +27,7 @@
  * @param <T> the upstream value type
  * @param <R> the downstream parameter type
  */
-public final class FlowableLift<R, T> extends FlowableWithUpstream<T, R> {
+public final class FlowableLift<R, T> extends AbstractFlowableWithUpstream<T, R> {
     /** The actual operator. */
     final FlowableOperator<? extends R, ? super T> operator;
 

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

@@ -20,7 +20,7 @@
 import io.reactivex.internal.fuseable.ConditionalSubscriber;
 import io.reactivex.internal.subscribers.flowable.*;
 
-public final class FlowableMap<T, U> extends FlowableWithUpstream<T, U> {
+public final class FlowableMap<T, U> extends AbstractFlowableWithUpstream<T, U> {
     final Function<? super T, ? extends U> mapper;
     public FlowableMap(Publisher<T> source, Function<? super T, ? extends U> mapper) {
         super(source);

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

@@ -22,7 +22,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableMapNotification<T, R> extends FlowableWithUpstream<T, R> {
+public final class FlowableMapNotification<T, R> extends AbstractFlowableWithUpstream<T, R> {
 
     final Function<? super T, ? extends R> onNextMapper;
     final Function<? super Throwable, ? extends R> onErrorMapper;

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

@@ -21,7 +21,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableMaterialize<T> extends FlowableWithUpstream<T, Try<Optional<T>>> {
+public final class FlowableMaterialize<T> extends AbstractFlowableWithUpstream<T, Try<Optional<T>>> {
 
     public FlowableMaterialize(Publisher<T> source) {
         super(source);

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

@@ -26,7 +26,7 @@
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableObserveOn<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableObserveOn<T> extends AbstractFlowableWithUpstream<T, T> {
 final Scheduler scheduler;
 
     final boolean delayError;

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

@@ -23,7 +23,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableOnBackpressureBuffer<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableOnBackpressureBuffer<T> extends AbstractFlowableWithUpstream<T, T> {
     final int bufferSize;
     final boolean unbounded;
     final boolean delayError;

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

@@ -21,7 +21,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableOnBackpressureDrop<T> extends FlowableWithUpstream<T, T> implements Consumer<T> {
+public final class FlowableOnBackpressureDrop<T> extends AbstractFlowableWithUpstream<T, T> implements Consumer<T> {
 
     final Consumer<? super T> onDrop;
 

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

@@ -20,7 +20,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 
-public final class FlowableOnBackpressureLatest<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableOnBackpressureLatest<T> extends AbstractFlowableWithUpstream<T, T> {
 
     public FlowableOnBackpressureLatest(Publisher<T> source) {
         super(source);

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

@@ -20,7 +20,7 @@
 import io.reactivex.internal.subscriptions.SubscriptionArbiter;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class FlowableOnErrorNext<T> extends FlowableWithUpstream<T, T> {
+public final class FlowableOnErrorNext<T> extends AbstractFlowableWithUpstream<T, T> {
     final Function<? super Throwable, ? extends Publisher<? extends T>> nextSupplier;
     final boolean allowFatal;