Add 'WithUpstream' interfaces to connectable implementations. (#4333)

Author: JakeWharton

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

@@ -33,9 +33,9 @@
  * manner. 
  * @param <T> the value type
  */
-public final class FlowablePublish<T> extends ConnectableFlowable<T> {
+public final class FlowablePublish<T> extends ConnectableFlowable<T> implements FlowableWithUpstream<T> {
     /** The source observable. */
-    final Publisher<? extends T> source;
+    final Publisher<T> source;
     /** Holds the current subscriber that is, will be or just was subscribed to the source observable. */
     final AtomicReference<PublishSubscriber<T>> current;
 
@@ -51,7 +51,7 @@ public final class FlowablePublish<T> extends ConnectableFlowable<T> {
      * @param bufferSize the size of the prefetch buffer
      * @return the connectable observable
      */
-    public static <T> ConnectableFlowable<T> create(Flowable<? extends T> source, final int bufferSize) {
+    public static <T> ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize) {
         // the current connection to source needs to be shared between the operator and its onSubscribe call
         final AtomicReference<PublishSubscriber<T>> curr = new AtomicReference<PublishSubscriber<T>>();
         Publisher<T> onSubscribe = new Publisher<T>() {
@@ -121,7 +121,7 @@ public void subscribe(Subscriber<? super T> child) {
         return new FlowablePublish<T>(onSubscribe, source, curr, bufferSize);
     }
 
-    public static <T, R> Flowable<R> create(final Flowable<? extends T> source, 
+    public static <T, R> Flowable<R> create(final Flowable<T> source,
             final Function<? super Flowable<T>, ? extends Publisher<R>> selector, final int bufferSize) {
         return unsafeCreate(new Publisher<R>() {
             @Override
@@ -151,14 +151,19 @@ public void accept(Disposable r) {
         });
     }
 
-    private FlowablePublish(Publisher<T> onSubscribe, Publisher<? extends T> source, 
+    private FlowablePublish(Publisher<T> onSubscribe, Publisher<T> source,
             final AtomicReference<PublishSubscriber<T>> current, int bufferSize) {
         this.onSubscribe = onSubscribe;
         this.source = source;
         this.current = current;
         this.bufferSize = bufferSize;
     }
 
+    @Override
+    public Publisher<T> source() {
+        return source;
+    }
+
     @Override
     protected void subscribeActual(Subscriber<? super T> s) {
         onSubscribe.subscribe(s);

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

@@ -30,9 +30,9 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Timed;
 
-public final class FlowableReplay<T> extends ConnectableFlowable<T> {
+public final class FlowableReplay<T> extends ConnectableFlowable<T> implements FlowableWithUpstream<T> {
     /** The source observable. */
-    final Publisher<? extends T> source;
+    final Publisher<T> source;
     /** Holds the current subscriber that is, will be or just was subscribed to the source observable. */
     final AtomicReference<ReplaySubscriber<T>> current;
     /** A factory that creates the appropriate buffer for the ReplaySubscriber. */
@@ -142,7 +142,7 @@ public static <T> ConnectableFlowable<T> createFrom(Flowable<? extends T> source
      * @param bufferSize
      * @return the new ConnectableObservable instance
      */
-    public static <T> ConnectableFlowable<T> create(Flowable<? extends T> source, 
+    public static <T> ConnectableFlowable<T> create(Flowable<T> source,
             final int bufferSize) {
         if (bufferSize == Integer.MAX_VALUE) {
             return createFrom(source);
@@ -164,7 +164,7 @@ public ReplayBuffer<T> call() {
      * @param scheduler
      * @return the new ConnectableObservable instance
      */
-    public static <T> ConnectableFlowable<T> create(Flowable<? extends T> source, 
+    public static <T> ConnectableFlowable<T> create(Flowable<T> source,
             long maxAge, TimeUnit unit, Scheduler scheduler) {
         return create(source, maxAge, unit, scheduler, Integer.MAX_VALUE);
     }
@@ -179,7 +179,7 @@ public static <T> ConnectableFlowable<T> create(Flowable<? extends T> source,
      * @param bufferSize
      * @return the new NbpConnectableObservable instance
      */
-    public static <T> ConnectableFlowable<T> create(Flowable<? extends T> source, 
+    public static <T> ConnectableFlowable<T> create(Flowable<T> source,
             final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize) {
         return create(source, new Callable<ReplayBuffer<T>>() {
             @Override
@@ -195,7 +195,7 @@ public ReplayBuffer<T> call() {
      * @param bufferFactory the factory to instantiate the appropriate buffer when the observable becomes active
      * @return the connectable observable
      */
-    static <T> ConnectableFlowable<T> create(Flowable<? extends T> source, 
+    static <T> ConnectableFlowable<T> create(Flowable<T> source,
             final Callable<? extends ReplayBuffer<T>> bufferFactory) {
         // the current connection to source needs to be shared between the operator and its onSubscribe call
         final AtomicReference<ReplaySubscriber<T>> curr = new AtomicReference<ReplaySubscriber<T>>();
@@ -248,7 +248,7 @@ public void subscribe(Subscriber<? super T> child) {
         return new FlowableReplay<T>(onSubscribe, source, curr, bufferFactory);
     }
 
-    private FlowableReplay(Publisher<T> onSubscribe, Flowable<? extends T> source, 
+    private FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
             final AtomicReference<ReplaySubscriber<T>> current,
             final Callable<? extends ReplayBuffer<T>> bufferFactory) {
         this.onSubscribe = onSubscribe;
@@ -257,6 +257,11 @@ private FlowableReplay(Publisher<T> onSubscribe, Flowable<? extends T> source,
         this.bufferFactory = bufferFactory;
     }
 
+    @Override
+    public Publisher<T> source() {
+        return source;
+    }
+
     @Override
     protected void subscribeActual(Subscriber<? super T> s) {
         onSubscribe.subscribe(s);

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

@@ -29,9 +29,9 @@
  * manner. 
  * @param <T> the value type
  */
-public final class ObservablePublish<T> extends ConnectableObservable<T> {
+public final class ObservablePublish<T> extends ConnectableObservable<T> implements ObservableWithUpstream<T> {
     /** The source observable. */
-    final ObservableSource<? extends T> source;
+    final ObservableSource<T> source;
     /** Holds the current subscriber that is, will be or just was subscribed to the source observable. */
     final AtomicReference<PublishSubscriber<T>> current;
 
@@ -47,7 +47,7 @@ public final class ObservablePublish<T> extends ConnectableObservable<T> {
      * @param bufferSize the size of the prefetch buffer
      * @return the connectable observable
      */
-    public static <T> ConnectableObservable<T> create(ObservableSource<? extends T> source, final int bufferSize) {
+    public static <T> ConnectableObservable<T> create(ObservableSource<T> source, final int bufferSize) {
         // the current connection to source needs to be shared between the operator and its onSubscribe call
         final AtomicReference<PublishSubscriber<T>> curr = new AtomicReference<PublishSubscriber<T>>();
         ObservableSource<T> onSubscribe = new ObservableSource<T>() {
@@ -117,7 +117,7 @@ public void subscribe(Observer<? super T> child) {
         return new ObservablePublish<T>(onSubscribe, source, curr, bufferSize);
     }
 
-    public static <T, R> Observable<R> create(final ObservableSource<? extends T> source,
+    public static <T, R> Observable<R> create(final ObservableSource<T> source,
                                               final Function<? super Observable<T>, ? extends ObservableSource<R>> selector, final int bufferSize) {
         return new Observable<R>() {
             @Override
@@ -148,14 +148,19 @@ public void accept(Disposable r) {
         };
     }
 
-    private ObservablePublish(ObservableSource<T> onSubscribe, ObservableSource<? extends T> source,
+    private ObservablePublish(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                               final AtomicReference<PublishSubscriber<T>> current, int bufferSize) {
         this.onSubscribe = onSubscribe;
         this.source = source;
         this.current = current;
         this.bufferSize = bufferSize;
     }
-    
+
+    @Override
+    public ObservableSource<T> source() {
+        return source;
+    }
+
     @Override
     protected void subscribeActual(Observer<? super T> observer) {
         onSubscribe.subscribe(observer);

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

@@ -28,9 +28,9 @@
 import io.reactivex.observables.ConnectableObservable;
 import io.reactivex.schedulers.Timed;
 
-public final class ObservableReplay<T> extends ConnectableObservable<T> {
+public final class ObservableReplay<T> extends ConnectableObservable<T> implements ObservableWithUpstream<T> {
     /** The source observable. */
-    final Observable<? extends T> source;
+    final Observable<T> source;
     /** Holds the current subscriber that is, will be or just was subscribed to the source observable. */
     final AtomicReference<ReplaySubscriber<T>> current;
     /** A factory that creates the appropriate buffer for the ReplaySubscriber. */
@@ -126,7 +126,7 @@ public static <T> ConnectableObservable<T> createFrom(Observable<? extends T> so
      * @param bufferSize
      * @return the new NbpConnectableObservable instance
      */
-    public static <T> ConnectableObservable<T> create(Observable<? extends T> source, 
+    public static <T> ConnectableObservable<T> create(Observable<T> source,
             final int bufferSize) {
         if (bufferSize == Integer.MAX_VALUE) {
             return createFrom(source);
@@ -148,7 +148,7 @@ public ReplayBuffer<T> call() {
      * @param scheduler
      * @return the new NbpConnectableObservable instance
      */
-    public static <T> ConnectableObservable<T> create(Observable<? extends T> source, 
+    public static <T> ConnectableObservable<T> create(Observable<T> source,
             long maxAge, TimeUnit unit, Scheduler scheduler) {
         return create(source, maxAge, unit, scheduler, Integer.MAX_VALUE);
     }
@@ -163,7 +163,7 @@ public static <T> ConnectableObservable<T> create(Observable<? extends T> source
      * @param bufferSize
      * @return the new NbpConnectableObservable instance
      */
-    public static <T> ConnectableObservable<T> create(Observable<? extends T> source, 
+    public static <T> ConnectableObservable<T> create(Observable<T> source,
             final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize) {
         return create(source, new Callable<ReplayBuffer<T>>() {
             @Override
@@ -179,7 +179,7 @@ public ReplayBuffer<T> call() {
      * @param bufferFactory the factory to instantiate the appropriate buffer when the observable becomes active
      * @return the connectable observable
      */
-    static <T> ConnectableObservable<T> create(Observable<? extends T> source, 
+    static <T> ConnectableObservable<T> create(Observable<T> source,
             final Callable<? extends ReplayBuffer<T>> bufferFactory) {
         // the current connection to source needs to be shared between the operator and its onSubscribe call
         final AtomicReference<ReplaySubscriber<T>> curr = new AtomicReference<ReplaySubscriber<T>>();
@@ -236,15 +236,20 @@ public void subscribe(Observer<? super T> child) {
         return new ObservableReplay<T>(onSubscribe, source, curr, bufferFactory);
     }
 
-    private ObservableReplay(ObservableSource<T> onSubscribe, Observable<? extends T> source,
+    private ObservableReplay(ObservableSource<T> onSubscribe, Observable<T> source,
                              final AtomicReference<ReplaySubscriber<T>> current,
                              final Callable<? extends ReplayBuffer<T>> bufferFactory) {
         this.onSubscribe = onSubscribe;
         this.source = source;
         this.current = current;
         this.bufferFactory = bufferFactory;
     }
-    
+
+    @Override
+    public ObservableSource<T> source() {
+        return source;
+    }
+
     @Override
     protected void subscribeActual(Observer<? super T> observer) {
         onSubscribe.subscribe(observer);