1.x: fix doOnRequest premature requesting.

When the operator is assembled and it sets the producer on the child, if
the child doesn't request until the assembly is complete, doOnRequest
requested the default Long.MAX_VALUE.

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,51 @@ 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);
+        
+        System.out.println(requested);
+        Assert.assertEquals(2, requested.size());
+        Assert.assertEquals(Arrays.asList(0L, 1L), requested);
+    }
 }