Merge pull request #2575 from abersnaze/subscribeOn_subscription

subscribeOn drops the subscriptions returned from the scheduler.

Author: akarnokd

src/main/java/rx/internal/operators/OperatorSubscribeOn.java

@@ -54,7 +54,7 @@ public void onError(Throwable e) {
 
             @Override
             public void onNext(final Observable<T> o) {
-                inner.schedule(new Action0() {
+                subscriber.add(inner.schedule(new Action0() {
 
                     @Override
                     public void call() {
@@ -102,7 +102,7 @@ public void call() {
 
                         });
                     }
-                });
+                }));
             }
 
         };

src/test/java/rx/internal/operators/OperatorReplayTest.java

@@ -609,6 +609,7 @@ public void testIssue2191_SchedulerUnsubscribe() throws Exception {
 
         verify(spiedWorker, times(1)).unsubscribe();
         verify(sourceUnsubscribed, times(1)).call();
+        verify(mockSubscription, times(1)).unsubscribe();
 
         verifyNoMoreInteractions(sourceNext);
         verifyNoMoreInteractions(sourceCompleted);
@@ -668,6 +669,7 @@ public void testIssue2191_SchedulerUnsubscribeOnError() throws Exception {
 
         verify(spiedWorker, times(1)).unsubscribe();
         verify(sourceUnsubscribed, times(1)).call();
+        verify(mockSubscription, times(1)).unsubscribe();
 
         verifyNoMoreInteractions(sourceNext);
         verifyNoMoreInteractions(sourceCompleted);