Merge pull request #2851 from zsxwing/fix-debounce

Add 'request(Long.MAX_VALUE)' in 'onStart' to fix the backpressure issue of debounce

Author: zsxwing

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

@@ -62,6 +62,12 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
         return new Subscriber<T>(child) {
             final DebounceState<T> state = new DebounceState<T>();
             final Subscriber<?> self = this;
+
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
+
             @Override
             public void onNext(final T t) {
 

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

@@ -23,6 +23,7 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Before;
@@ -36,6 +37,7 @@
 import rx.exceptions.TestException;
 import rx.functions.Action0;
 import rx.functions.Func1;
+import rx.observers.TestSubscriber;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 
@@ -287,4 +289,20 @@ public Observable<Integer> call(Integer t1) {
         verify(o).onCompleted();
         verify(o, never()).onError(any(Throwable.class));
     }
+
+    @Test
+    public void debounceWithTimeBackpressure() throws InterruptedException {
+        TestScheduler scheduler = new TestScheduler();
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        Observable.merge(
+                Observable.just(1),
+                Observable.just(2).delay(10, TimeUnit.MILLISECONDS, scheduler)
+        ).debounce(20, TimeUnit.MILLISECONDS, scheduler).take(1).subscribe(subscriber);
+
+        scheduler.advanceTimeBy(30, TimeUnit.MILLISECONDS);
+
+        subscriber.assertReceivedOnNext(Arrays.asList(2));
+        subscriber.assertTerminalEvent();
+        subscriber.assertNoErrors();
+    }
 }