Merge pull request #2820 from akarnokd/WindowWithSizeBackpressure

Backpressure for window(size)

Author: benjchristensen

src/main/java/rx/Observable.java

@@ -8968,7 +8968,8 @@ public final <TClosing> Observable<Observable<T>> window(Func0<? extends Observa
      * <img width="640" height="400" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/window3.png" alt="">
      * <dl>
      *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>This operator does not support backpressure as it uses {@code count} to control data flow.</dd>
+     *  <dd>The operator honors backpressure on its outer subscriber, ignores backpressure in its inner Observables 
+     *  but each of them will emit at most {@code count} elements.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>This version of {@code window} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
@@ -8992,7 +8993,8 @@ public final Observable<Observable<T>> window(int count) {
      * <img width="640" height="365" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/window4.png" alt="">
      * <dl>
      *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>This operator does not support backpressure as it uses {@code count} to control data flow.</dd>
+     *  <dd>The operator has limited backpressure support. If {@code count} == {@code skip}, the operator honors backpressure on its outer subscriber, ignores backpressure in its inner Observables 
+     *  but each of them will emit at most {@code count} elements.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>This version of {@code window} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>

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

@@ -15,18 +15,14 @@
  */
 package rx.internal.operators;
 
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
+import java.util.*;
 
-import rx.Observable;
+import rx.*;
 import rx.Observable.Operator;
-import rx.Subscription;
+import rx.Observable;
+import rx.Observer;
 import rx.functions.Action0;
 import rx.subscriptions.Subscriptions;
-import rx.Observer;
-import rx.Subscriber;
 
 /**
  * Creates windows of values into the source sequence with skip frequency and size bounds.
@@ -78,26 +74,36 @@ public ExactSubscriber(Subscriber<? super Observable<T>> child) {
                 @Override
                 public void call() {
                     // if no window we unsubscribe up otherwise wait until window ends
-                    if(noWindow) {
+                    if (noWindow) {
                         parentSubscription.unsubscribe();
                     }
                 }
 
             }));
-        }
-
-        @Override
-        public void onStart() {
-            // no backpressure as we are controlling data flow by window size
-            request(Long.MAX_VALUE);
+            child.setProducer(new Producer() {
+                @Override
+                public void request(long n) {
+                    if (n > 0) {
+                        long u = n * size;
+                        if (((u >>> 31) != 0) && (u / n != size)) {
+                            u = Long.MAX_VALUE;
+                        }
+                        requestMore(u);
+                    }
+                }
+            });
         }
 
+        void requestMore(long n) {
+            request(n);
+        }
+
         @Override
         public void onNext(T t) {
             if (window == null) {
                 noWindow = false;
                 window = BufferUntilSubscriber.create();
-                child.onNext(window);                
+                child.onNext(window);
             }
             window.onNext(t);
             if (++count % size == 0) {

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

@@ -15,20 +15,19 @@
  */
 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.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
+import static org.mockito.Mockito.*;
+import rx.*;
 import rx.Observable;
-import rx.functions.Action1;
-import rx.functions.Func1;
+import rx.Observer;
+import rx.functions.*;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
 
@@ -198,5 +197,52 @@ private List<String> list(String... args) {
         }
         return list;
     }
+    
+    @Test
+    public void testBackpressureOuter() {
+        Observable<Observable<Integer>> source = Observable.range(1, 10).window(3);
+        
+        final List<Integer> list = new ArrayList<Integer>();
+        
+        @SuppressWarnings("unchecked")
+        final Observer<Integer> o = mock(Observer.class);
+        
+        source.subscribe(new Subscriber<Observable<Integer>>() {
+            @Override
+            public void onStart() {
+                request(1);
+            }
+            @Override
+            public void onNext(Observable<Integer> t) {
+                t.subscribe(new Observer<Integer>() {
+                    @Override
+                    public void onNext(Integer t) {
+                        list.add(t);
+                    }
+                    @Override
+                    public void onError(Throwable e) {
+                        o.onError(e);
+                    }
+                    @Override
+                    public void onCompleted() {
+                        o.onCompleted();
+                    }
+                });
+            }
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
+        });
+        
+        assertEquals(Arrays.asList(1, 2, 3), list);
+        
+        verify(o, never()).onError(any(Throwable.class));
+        verify(o, times(1)).onCompleted(); // 1 inner
+    }
 
 }