fix for #5429 (#5430)

Author: yurgisbaykshtis

src/main/java/rx/observables/AsyncOnSubscribe.java

@@ -532,8 +532,11 @@ boolean tryEmit(long n) {
                 onNextCalled = false;
                 expectedDelivery = n;
                 nextIteration(n);
-
-                if (hasTerminated || isUnsubscribed()) {
+                
+                //hasTerminated will be true when onCompleted was already emitted from the request callback 
+                //even if the the observer has not seen onCompleted from the requested observable, 
+                //so we should not clean up while there are active subscriptions
+                if (hasTerminated && !subscriptions.hasSubscriptions() || isUnsubscribed()) {
                     cleanup();
                     return true;
                 }

src/test/java/rx/observables/AsyncOnSubscribeTest.java

@@ -483,4 +483,38 @@ public Integer call(Integer state, Long requested, Observer<Observable<? extends
 
         subscriber.assertNotCompleted();
     }
+
+    @Test
+    public void testMergeDelayedWithScalar() {
+        final TestScheduler scheduler = new TestScheduler();
+        Observable<Integer> os = Observable.create(AsyncOnSubscribe.<Integer, Integer> createStateful(
+            new Func0<Integer>() {
+
+                @Override
+                public Integer call() {
+                    return 0;
+                }
+
+            },
+            new Func3<Integer, Long, Observer<Observable<? extends Integer>>, Integer>() {
+
+                @Override
+                public Integer call(Integer state, Long requested, Observer<Observable<? extends Integer>> emitter) {
+                    if (state == 0) {
+                        emitter.onNext(Observable.range(0,100).delay(1, TimeUnit.SECONDS, scheduler));
+                    } else {
+                        emitter.onCompleted();
+                    }
+                    return state + 1;
+                }
+
+            }));
+
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+        os.mergeWith(Observable.just(0)).subscribe(ts);
+        scheduler.advanceTimeBy(1, TimeUnit.HOURS);
+        ts.assertCompleted();
+        ts.assertValueCount(101);
+    }
+
 }