Merge pull request #2566 from akarnokd/CombineLatestBackpressureFix

CombineLatest: fixed concurrent requestUpTo yielding -1 requests

Author: akarnokd

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

@@ -230,9 +230,14 @@ public MultiSourceRequestableSubscriber(int index, int initial, Subscriber<? sup
         }
 
         public void requestUpTo(long n) {
-            long r = Math.min(emitted.get(), n);
-            request(r);
-            emitted.addAndGet(-r);
+            do {
+                long r = emitted.get();
+                long u = Math.min(r, n);
+                if (emitted.compareAndSet(r, r - u)) {
+                    request(u);
+                    break;
+                }
+            } while (true);
         }
 
         @Override

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

@@ -789,6 +789,13 @@ public Object call(Object... args) {
 
     }
 
+    @Test
+    public void testBackpressureLoop() {
+        for (int i = 0; i < 5000; i++) {
+            testBackpressure();
+        }
+    }
+    
     @Test
     public void testBackpressure() {
         Func2<String, Integer, String> combineLatestFunction = getConcatStringIntegerCombineLatestFunction();