Merge pull request #3727 from davidmoten/scan-request-bug

scan should pass upstream a request of Long.MAX_VALUE
ReactiveX · Feb 24, 2016 · a57bccc · a57bccc
2 parents 3e6affd + 457533c
commit a57bccc
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 1 deletion.
diff --git a/src/main/java/rx/internal/operators/OperatorScan.java b/src/main/java/rx/internal/operators/OperatorScan.java
@@ -268,8 +268,12 @@ public void setProducer(Producer p) {
                 if (producer != null) {
                     throw new IllegalStateException("Can't set more than one Producer!");
                 }
+                mr = missedRequested;
                 // request one less because of the initial value, this happens once
-                mr = missedRequested - 1;
+                // and is performed only if the request is not at MAX_VALUE already
+                if (mr != Long.MAX_VALUE) {
+                    mr -= 1;
+                }
                 missedRequested = 0L;
                 producer = p;
             }

diff --git a/src/test/java/rx/internal/operators/OperatorScanTest.java b/src/test/java/rx/internal/operators/OperatorScanTest.java
@@ -451,4 +451,22 @@ public void onNext(Integer t) {
             }
         });
     }
+
+    @Test
+    public void scanShouldPassUpstreamARequestForMaxValue() {
+        final List<Long> requests = new ArrayList<Long>();
+        Observable.just(1,2,3).doOnRequest(new Action1<Long>() {
+            @Override
+            public void call(Long n) {
+                requests.add(n);
+            }
+        })
+        .scan(new Func2<Integer,Integer, Integer>() {
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return 0;
+            }}).count().subscribe();
+
+        assertEquals(Arrays.asList(Long.MAX_VALUE), requests);
+    }
 }