1.x: fix filter() default-requesting and thus going unbounded (#3912)

Author: akarnokd

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

@@ -53,6 +53,7 @@ static final class FilterSubscriber<T> extends Subscriber<T> {
         public FilterSubscriber(Subscriber<? super T> actual, Func1<? super T, Boolean> predicate) {
             this.actual = actual;
             this.predicate = predicate;
+            request(0);
         }
 
         @Override

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

@@ -26,7 +26,7 @@
 import rx.*;
 import rx.exceptions.*;
 import rx.functions.*;
-import rx.internal.util.RxRingBuffer;
+import rx.internal.util.*;
 import rx.observers.TestSubscriber;
 import rx.subjects.PublishSubject;
 
@@ -69,7 +69,7 @@ public Boolean call(String t1) {
         });
 
         final CountDownLatch latch = new CountDownLatch(1);
-        TestSubscriber<String> ts = new TestSubscriber<String>() {
+        TestSubscriber<String> ts = new TestSubscriber<String>(0L) {
 
             @Override
             public void onCompleted() {
@@ -116,7 +116,7 @@ public Boolean call(Integer t1) {
         });
 
         final CountDownLatch latch = new CountDownLatch(1);
-        final TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+        final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L) {
 
             @Override
             public void onCompleted() {
@@ -194,4 +194,18 @@ public Boolean call(Integer v) {
         ts.assertError(TestException.class);
     }
 
+    @Test
+    public void doesntRequestOnItsOwn() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0L);
+        
+        Observable.range(1, 10).filter(UtilityFunctions.alwaysTrue()).unsafeSubscribe(ts);
+        
+        ts.assertNoValues();
+        
+        ts.requestMore(10);
+        
+        ts.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
 }