1.x: fix doOnRequest premature requesting

Author: akarnokd

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

@@ -57,6 +57,7 @@ private static final class ParentSubscriber<T> extends Subscriber<T> {
 
         ParentSubscriber(Subscriber<? super T> child) {
             this.child = child;
+            this.request(0);
         }
 
         private void requestMore(long n) {

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

@@ -1,19 +1,16 @@
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.*;
+import java.util.concurrent.atomic.*;
 
-import org.junit.Test;
+import org.junit.*;
 
+import rx.*;
 import rx.Observable;
-import rx.Subscriber;
-import rx.functions.Action0;
-import rx.functions.Action1;
+import rx.Observable.OnSubscribe;
+import rx.functions.*;
 
 public class OperatorDoOnRequestTest {
 
@@ -76,5 +73,55 @@ public void onNext(Integer t) {
                 });
         assertEquals(Arrays.asList(3L,1L,2L,3L,4L,5L), requests);
     }
+    
+    @Test
+    public void dontRequestIfDownstreamRequestsLate() {
+        final List<Long> requested = new ArrayList<Long>();
+
+        Action1<Long> empty = Actions.empty();
+        
+        final AtomicReference<Producer> producer = new AtomicReference<Producer>();
+        
+        Observable.create(new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> t) {
+                t.setProducer(new Producer() {
+                    @Override
+                    public void request(long n) {
+                        requested.add(n);
+                    }
+                });
+            }
+        }).doOnRequest(empty).subscribe(new Subscriber<Object>() {
+            @Override
+            public void onNext(Object t) {
+                
+            }
+            
+            @Override
+            public void onError(Throwable e) {
+                
+            }
+            
+            @Override
+            public void onCompleted() {
+                
+            }
+            
+            @Override
+            public void setProducer(Producer p) {
+                producer.set(p);
+            }
+        });
+        
+        producer.get().request(1);
 
+        int s = requested.size();
+        if (s == 1) {
+            // this allows for an implementation that itself doesn't request
+            Assert.assertEquals(Arrays.asList(1L), requested);
+        } else {
+            Assert.assertEquals(Arrays.asList(0L, 1L), requested);
+        }
+    }
 }