Fix Merge Bug

If the queue being drained did not complete due to the backpressure request being fulfilled, it could onComplete before draining the queue and miss values.

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/operators/OperatorMerge.java

@@ -535,7 +535,9 @@ private void emit(T t, boolean complete) {
                             emitted++;
                         }
                     } else {
-                        if (producer.requested > 0) {
+                        // this needs to check q.count() as draining above may not have drained the full queue
+                        // perf tests show this to be okay, though different queue implementations could perform poorly with this
+                        if (producer.requested > 0 && q.count() == 0) {
                             if (complete) {
                                 parentSubscriber.completeInner(this);
                             } else {