Merge pull request #2630 from akarnokd/FixSubscriberLoss

Cast back Observer to Subscriber if passed to subscribe(Observer)

Author: akarnokd

src/main/java/rx/Observable.java

@@ -7289,6 +7289,9 @@ public final void onNext(T args) {
      * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
      */
     public final Subscription subscribe(final Observer<? super T> observer) {
+        if (observer instanceof Subscriber) {
+            return subscribe((Subscriber<? super T>)observer);
+        }
         return subscribe(new Subscriber<T>() {
 
             @Override

src/test/java/rx/ObservableTests.java

@@ -1129,4 +1129,13 @@ public void testEmptyIsEmpty() {
         verify(w, never()).onNext(any(Integer.class));
         verify(w, never()).onError(any(Throwable.class));
     }
+    
+    @Test // cf. https://github.com/ReactiveX/RxJava/issues/2599
+    public void testSubscribingSubscriberAsObserverMaintainsSubscriptionChain() {
+        TestSubscriber<Object> subscriber = new TestSubscriber<Object>();
+        Subscription subscription = Observable.just("event").subscribe((Observer<Object>) subscriber);
+        subscription.unsubscribe();
+
+        subscriber.assertUnsubscribed();
+    }
 }