Backpressure for window(size)

Author: akarnokd

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
+    }
 
 }