2.x: fix window(time, size) not completing windows on timeout (ReactiveX#5106)

Author: akarnokd

Diff for: src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowTimed.java

@@ -496,6 +496,8 @@ void drainLoop() {
                     if (isHolder) {
                         ConsumerIndexHolder consumerIndexHolder = (ConsumerIndexHolder) o;
                         if (producerIndex == consumerIndexHolder.index) {
+                            w.onComplete();
+
                             w = UnicastProcessor.<T>create(bufferSize);
                             window = w;
 

Diff for: src/main/java/io/reactivex/internal/operators/observable/ObservableWindowTimed.java

@@ -439,6 +439,8 @@ void drainLoop() {
                     if (isHolder) {
                         ConsumerIndexHolder consumerIndexHolder = (ConsumerIndexHolder) o;
                         if (producerIndex == consumerIndexHolder.index) {
+                            w.onComplete();
+
                             w = UnicastSubject.create(bufferSize);
                             window = w;
 

Diff for: src/test/java/io/reactivex/internal/operators/flowable/FlowableWindowWithTimeTest.java

@@ -686,4 +686,22 @@ public Flowable<Integer> apply(Flowable<Integer> v) throws Exception {
         .awaitDone(1, TimeUnit.SECONDS)
         .assertResult(1, 2);
     }
+
+    @Test
+    public void sizeTimeTimeout() {
+        TestScheduler scheduler = new TestScheduler();
+        PublishProcessor<Integer> ps = PublishProcessor.<Integer>create();
+
+        TestSubscriber<Flowable<Integer>> ts = ps.window(5, TimeUnit.MILLISECONDS, scheduler, 100)
+        .test()
+        .assertValueCount(1);
+
+        scheduler.advanceTimeBy(5, TimeUnit.MILLISECONDS);
+
+        ts.assertValueCount(2)
+        .assertNoErrors()
+        .assertNotComplete();
+
+        ts.values().get(0).test().assertResult();
+    }
 }

Diff for: src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithTimeTest.java

@@ -585,4 +585,22 @@ public ObservableSource<Integer> apply(Observable<Integer> v) throws Exception {
         .awaitDone(1, TimeUnit.SECONDS)
         .assertResult(1, 2);
     }
+
+    @Test
+    public void sizeTimeTimeout() {
+        TestScheduler scheduler = new TestScheduler();
+        Subject<Integer> ps = PublishSubject.<Integer>create();
+
+        TestObserver<Observable<Integer>> ts = ps.window(5, TimeUnit.MILLISECONDS, scheduler, 100)
+        .test()
+        .assertValueCount(1);
+
+        scheduler.advanceTimeBy(5, TimeUnit.MILLISECONDS);
+
+        ts.assertValueCount(2)
+        .assertNoErrors()
+        .assertNotComplete();
+
+        ts.values().get(0).test().assertResult();
+    }
 }