Merge pull request #3167 from akarnokd/ReplayManageRequestsFix

Fixed negative request due to unsubscription of a large requester

Author: akarnokd

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

@@ -501,7 +501,7 @@ void manageRequests() {
                 InnerProducer<T>[] a = producers.get();
 
                 long ri = maxChildRequested;
-                long maxTotalRequests = 0;
+                long maxTotalRequests = ri;
 
                 for (InnerProducer<T> rp : a) {
                     maxTotalRequests = Math.max(maxTotalRequests, rp.totalRequested.get());

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

@@ -1120,4 +1120,29 @@ public void onNext(Integer t) {
         ts.assertNotCompleted();
         ts.assertError(TestException.class);
     }
+    
+    @Test
+    public void unboundedLeavesEarly() {
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        final List<Long> requests = new ArrayList<Long>();
+
+        Observable<Integer> out = source
+                .doOnRequest(new Action1<Long>() {
+                    @Override
+                    public void call(Long t) {
+                        requests.add(t);
+                    }
+                }).replay().autoConnect();
+        
+        TestSubscriber<Integer> ts1 = TestSubscriber.create(5);
+        TestSubscriber<Integer> ts2 = TestSubscriber.create(10);
+        
+        out.subscribe(ts1);
+        out.subscribe(ts2);
+        ts2.unsubscribe();
+        
+        Assert.assertEquals(Arrays.asList(5L, 5L), requests);
+    }
+    
 }