Concat Outer Backpressure

Fixed #1481

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/operators/OperatorConcat.java

@@ -63,7 +63,14 @@ public void call() {
                 }
             }));
         }
-        
+
+        @Override
+        public void onStart() {
+            // no need for more than 1 at a time since we concat 1 at a time, so we'll request 2 to start ...
+            // 1 to be subscribed to, 1 in the queue, then we'll keep requesting 1 at a time after that
+            request(2);
+        }
+
         @Override
         public void onNext(Observable<? extends T> t) {
             queue.add(nl.next(t));
@@ -85,22 +92,25 @@ public void onCompleted() {
                 subscribeNext();
             }
         }
+
         void completeInner() {
+            request(1);
             if (WIP_UPDATER.decrementAndGet(this) > 0) {
                 subscribeNext();
             }
         }
+
         void subscribeNext() {
             Object o = queue.poll();
             if (nl.isCompleted(o)) {
                 s.onCompleted();
-            } else 
-            if (o != null) {
+            } else if (o != null) {
                 Observable<? extends T> obs = nl.getValue(o);
                 Subscriber<T> sourceSub = new Subscriber<T>() {
 
                     @Override
                     public void onNext(T t) {
+                        // TODO need to support backpressure here https://github.com/Netflix/RxJava/issues/1480
                         s.onNext(t);
                     }
 

rxjava-core/src/main/java/rx/internal/operators/OperatorWindowWithSize.java

@@ -63,6 +63,12 @@ public ExactSubscriber(Subscriber<? super Observable<T>> child) {
             this.child = child;
         }
 
+        @Override
+        public void onStart() {
+            // no backpressure as we are controlling data flow by window size
+            request(Long.MAX_VALUE);
+        }
+        
         @Override
         public void onNext(T t) {
             if (count++ % size == 0) {
@@ -106,6 +112,12 @@ public InexactSubscriber(Subscriber<? super Observable<T>> child) {
             this.chunks = new LinkedList<CountedSubject<T>>();
         }
 
+        @Override
+        public void onStart() {
+            // no backpressure as we are controlling data flow by window size
+            request(Long.MAX_VALUE);
+        }
+        
         @Override
         public void onNext(T t) {
             if (count++ % skip == 0) {

rxjava-core/src/test/java/rx/internal/operators/OperatorConcatTest.java

@@ -15,6 +15,7 @@
  */
 package rx.internal.operators;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
@@ -650,4 +651,13 @@ public void call(Subscriber<? super Observable<Integer>> s) {
         inOrder.verify(o).onCompleted();
         verify(o, never()).onError(any(Throwable.class));
     }
+    
+    @Test
+    public void testConcatOuterBackpressure() {
+        assertEquals(1,
+                (int) Observable.<Integer> empty()
+                        .concatWith(Observable.just(1))
+                        .take(1)
+                        .toBlocking().single());
+    }
 }