match bug fix 2

davidmoten · Oct 26, 2016 · 1263c8a · 1263c8a
1 parent 56c5248
commit 1263c8a
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 12 deletions.
diff --git a/src/main/java/com/github/davidmoten/rx/internal/operators/OnSubscribeMatch.java b/src/main/java/com/github/davidmoten/rx/internal/operators/OnSubscribeMatch.java
@@ -231,18 +231,7 @@ private int handleItem(Object value, Source source) {
                 }
             }
             // requests are batched so that each source gets a turn
-            if (requestFromA == requestSize && completed == COMPLETED_B) {
-                requestFromA = 0;
-                aSub.requestMore(requestSize);
-            } else if (requestFromB == requestSize && completed == COMPLETED_A) {
-                requestFromB = 0;
-                bSub.requestMore(requestSize);
-            } else if (requestFromA == requestSize && requestFromB == requestSize) {
-                requestFromA = 0;
-                requestFromB = 0;
-                aSub.requestMore(requestSize);
-                bSub.requestMore(requestSize);
-            }
+            checkToRequestMore();
             return numEmitted;
         }
 
@@ -259,6 +248,23 @@ private void handleCompleted(CompletedFrom comp) {
             if (done) {
                 clear();
                 child.onCompleted();
+            } else {
+                checkToRequestMore();
+            }
+        }
+
+        private void checkToRequestMore() {
+            if (requestFromA == requestSize && completed == COMPLETED_B) {
+                requestFromA = 0;
+                aSub.requestMore(requestSize);
+            } else if (requestFromB == requestSize && completed == COMPLETED_A) {
+                requestFromB = 0;
+                bSub.requestMore(requestSize);
+            } else if (requestFromA == requestSize && requestFromB == requestSize) {
+                requestFromA = 0;
+                requestFromB = 0;
+                aSub.requestMore(requestSize);
+                bSub.requestMore(requestSize);
             }
         }
 

diff --git a/src/test/java/com/github/davidmoten/rx/internal/operators/OnSubscribeMatchTest.java b/src/test/java/com/github/davidmoten/rx/internal/operators/OnSubscribeMatchTest.java
@@ -49,6 +49,13 @@ public void testKeepsRequesting() {
         Observable<Integer> b = Observable.just(2).repeat(1000).concatWith(Observable.just(1));
         match(a, b, 1);
     }
+
+    @Test
+    public void testKeepsRequestingSwitched() {
+        Observable<Integer> a = Observable.just(2).repeat(1000).concatWith(Observable.just(1));
+        Observable<Integer> b = Observable.just(1);
+        match(a, b, 1);
+    }
 
     @Test
     public void test2() {
@@ -98,6 +105,13 @@ public void testNoMatchExistsForAtLeastOneFirstLonger() {
         Observable<Integer> b = Observable.just(1);
         match(a, b, 1);
     }
+
+    @Test
+    public void testNoMatchExistsForAtLeastOneFirstLongerSwitched() {
+        Observable<Integer> a = Observable.just(1);
+        Observable<Integer> b = Observable.just(1, 2);
+        match(a, b, 1);
+    }
 
     @Test
     public void testNoMatchExistsForAtLeastOneSameLength() {