From 20430945276d8522dc3fe3e16463330b1a311c1c Mon Sep 17 00:00:00 2001
From: Dave Moten <davidmoten@gmail.com>
Date: Tue, 24 Feb 2015 09:06:43 +1100
Subject: [PATCH] add request overflow check for combineLatest

---
 .../operators/OnSubscribeCombineLatest.java   |  2 +-
 .../OnSubscribeCombineLatestTest.java         | 35 +++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java b/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java
index c5ea0d1291..1537f91aaf 100644
--- a/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java
+++ b/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java
@@ -110,7 +110,7 @@ public MultiSourceProducer(final Subscriber<? super R> child, final List<? exten
 
         @Override
         public void request(long n) {
-            requested.getAndAdd(n);
+            BackpressureUtils.getAndAddRequest(requested, n);
             if (!started.get() && started.compareAndSet(false, true)) {
                 /*
                  * NOTE: this logic will ONLY work if we don't have more sources than the size of the buffer.
diff --git a/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java b/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java
index d3de0c0186..c2a9f15dce 100644
--- a/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java
+++ b/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java
@@ -851,5 +851,40 @@ public Long call(Long t1, Integer t2) {
 
         assertEquals(SIZE, count.get());
     }
+    
+    @Test(timeout=10000)
+    public void testCombineLatestRequestOverflow() throws InterruptedException {
+        List<Observable<Integer>> sources = Arrays.asList(Observable.from(Arrays.asList(1,2,3,4)), Observable.from(Arrays.asList(5,6,7,8)));
+        Observable<Integer> o = Observable.combineLatest(sources,new FuncN<Integer>() {
+            @Override
+            public Integer call(Object... args) {
+               return (Integer) args[0];
+            }});
+        //should get at least 4
+        final CountDownLatch latch = new CountDownLatch(4);
+        o.subscribeOn(Schedulers.computation()).subscribe(new Subscriber<Integer>() {
+            
+            @Override
+            public void onStart() {
+                request(2);
+            }
+
+            @Override
+            public void onCompleted() {
+                //ignore
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                throw new RuntimeException(e);
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                latch.countDown();
+                request(Long.MAX_VALUE-1);
+            }});
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+    }
 
 }