Merge pull request ReactiveX#1904 from zsxwing/fix-scan

Fix the bug that Scan may request 0 when n is 1

Author: benjchristensen

src/main/java/rx/internal/operators/OperatorScan.java

@@ -140,17 +140,30 @@ public void setProducer(final Producer producer) {
 
                     final AtomicBoolean once = new AtomicBoolean();
 
+                    final AtomicBoolean excessive = new AtomicBoolean();
+
                     @Override
                     public void request(long n) {
                         if (once.compareAndSet(false, true)) {
                             if (initialValue == NO_INITIAL_VALUE || n == Long.MAX_VALUE) {
                                 producer.request(n);
                             } else {
-                                producer.request(n - 1);
+                                if (n == Long.MAX_VALUE) {
+                                    producer.request(Long.MAX_VALUE);
+                                } else if (n == 1) {
+                                    excessive.set(true);
+                                    producer.request(1); // request at least 1
+                                } else {
+                                    producer.request(n - 1);
+                                }
                             }
                         } else {
                             // pass-thru after first time
-                            producer.request(n);
+                            if (excessive.compareAndSet(true, false) && n != Long.MAX_VALUE) {
+                                producer.request(n - 1);
+                            } else {
+                                producer.request(n);
+                            }
                         }
                     }
                 });

src/test/java/rx/internal/operators/OperatorScanTest.java

@@ -295,4 +295,21 @@ public void call(List<Integer> list, Integer t2) {
         assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), o.toBlocking().single());
         assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), o.toBlocking().single());
     }
+
+    @Test
+    public void testScanWithRequestOne() {
+        Observable<Integer> o = Observable.just(1, 2).scan(0, new Func2<Integer, Integer, Integer>() {
+
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return t1 + t2;
+            }
+
+        }).take(1);
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        o.subscribe(subscriber);
+        subscriber.assertReceivedOnNext(Arrays.asList(0));
+        subscriber.assertTerminalEvent();
+        subscriber.assertNoErrors();
+    }
 }