Fixes Early Unsubscribe for Overlapping Window

- #1546
- This also fixes the fact that the overlapping window overload was not propagating unsubscribe before. A new unit test caught that.

Author: benjchristensen

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

@@ -124,14 +124,35 @@ public void onCompleted() {
             child.onCompleted();
         }
     }
+
     /** Subscriber with inexact, possibly overlapping or skipping windows. */
     final class InexactSubscriber extends Subscriber<T> {
         final Subscriber<? super Observable<T>> child;
         int count;
-        final List<CountedSubject<T>> chunks;
+        final List<CountedSubject<T>> chunks = new LinkedList<CountedSubject<T>>();
+        Subscription parentSubscription = this;
+
         public InexactSubscriber(Subscriber<? super Observable<T>> child) {
+            /**
+             * See https://github.com/ReactiveX/RxJava/issues/1546
+             * We cannot compose through a Subscription because unsubscribing
+             * applies to the outer, not the inner.
+             */
             this.child = child;
-            this.chunks = new LinkedList<CountedSubject<T>>();
+            /*
+             * Add unsubscribe hook to child to get unsubscribe on outer (unsubscribing on next window, not on the inner window itself)
+             */
+            child.add(Subscriptions.create(new Action0() {
+
+                @Override
+                public void call() {
+                    // if no window we unsubscribe up otherwise wait until window ends
+                    if (chunks == null || chunks.size() == 0) {
+                        parentSubscription.unsubscribe();
+                    }
+                }
+
+            }));
         }
 
         @Override
@@ -143,10 +164,13 @@ public void onStart() {
         @Override
         public void onNext(T t) {
             if (count++ % skip == 0) {
-                CountedSubject<T> cs = createCountedSubject();
-                chunks.add(cs);
-                child.onNext(cs.producer);
+                if (!child.isUnsubscribed()) {
+                    CountedSubject<T> cs = createCountedSubject();
+                    chunks.add(cs);
+                    child.onNext(cs.producer);
+                }
             }
+
             Iterator<CountedSubject<T>> it = chunks.iterator();
             while (it.hasNext()) {
                 CountedSubject<T> cs = it.next();
@@ -156,6 +180,10 @@ public void onNext(T t) {
                     cs.consumer.onCompleted();
                 }
             }
+            if (chunks.size() == 0 && child.isUnsubscribed()) {
+                parentSubscription.unsubscribe();
+                return;
+            }
         }
 
         @Override
@@ -177,6 +205,7 @@ public void onCompleted() {
             }
             child.onCompleted();
         }
+
         CountedSubject<T> createCountedSubject() {
             final BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
             return new CountedSubject<T>(bus, bus);

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

@@ -148,6 +148,49 @@ public void call(Integer t1) {
         assertTrue(count.get() < 100000);
     }
 
+    @Test
+    public void testWindowUnsubscribeOverlapping() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        final AtomicInteger count = new AtomicInteger();
+        Observable.merge(Observable.range(1, 10000).doOnNext(new Action1<Integer>() {
+
+            @Override
+            public void call(Integer t1) {
+                count.incrementAndGet();
+            }
+
+        }).window(5, 4).take(2)).subscribe(ts);
+        ts.awaitTerminalEvent(500, TimeUnit.MILLISECONDS);
+        ts.assertTerminalEvent();
+        //        System.out.println(ts.getOnNextEvents());
+        ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 5, 6, 7, 8, 9));
+        assertEquals(9, count.get());
+    }
+
+    @Test
+    public void testWindowUnsubscribeOverlappingAsyncSource() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        final AtomicInteger count = new AtomicInteger();
+        Observable.merge(Observable.range(1, 100000)
+                .doOnNext(new Action1<Integer>() {
+
+                    @Override
+                    public void call(Integer t1) {
+                        count.incrementAndGet();
+                    }
+
+                })
+                .observeOn(Schedulers.computation())
+                .window(5, 4)
+                .take(2))
+                .subscribe(ts);
+        ts.awaitTerminalEvent(500, TimeUnit.MILLISECONDS);
+        ts.assertTerminalEvent();
+        ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 5, 6, 7, 8, 9));
+        // make sure we don't emit all values ... the unsubscribe should propagate
+        assertTrue(count.get() < 100000);
+    }
+
     private List<String> list(String... args) {
         List<String> list = new ArrayList<String>();
         for (String arg : args) {