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;