Merge pull request #928 from akarnokd/SubscribeOnBoundedFix

Fix deadlock in SubscribeOnBounded

Author: benjchristensen

rxjava-core/src/main/java/rx/operators/OperatorSubscribeOnBounded.java

@@ -88,14 +88,13 @@ public void onNext(final Observable<T> o) {
                 if (checkNeedBuffer(o)) {
                     // use buffering (possibly blocking) for a possibly synchronous subscribe
                     final BufferUntilSubscriber<T> bus = new BufferUntilSubscriber<T>(bufferSize, subscriber);
-                    o.subscribe(bus);
                     subscriber.add(scheduler.schedule(new Action1<Inner>() {
                         @Override
                         public void call(final Inner inner) {
                             bus.enterPassthroughMode();
                         }
                     }));
-                    return;
+                    o.subscribe(bus);
                 } else {
                     // no buffering (async subscribe)
                     subscriber.add(scheduler.schedule(new Action1<Inner>() {

rxjava-core/src/test/java/rx/operators/OperatorSubscribeOnBoundedTest.java

@@ -399,5 +399,18 @@ public void call(Subscriber<? super Integer> sub) {
         ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
         assertEquals(10, count.get());
     }
-
+    @Test(timeout = 2000)
+    public void testNoDeadlock() {
+        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
+        Observable<Integer> source = Observable.from(data);
+        
+        Observable<Integer> result = source.nest().lift(new OperatorSubscribeOnBounded<Integer>(Schedulers.newThread(), 1));
+        
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        
+        result.subscribe(ts);
+        
+        ts.awaitTerminalEvent();
+        ts.assertReceivedOnNext(data);
+    }
 }