diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java index d7c3e66e3d..0f1b43ae4a 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java @@ -498,7 +498,7 @@ void drainLoop() { if (isHolder) { ConsumerIndexHolder consumerIndexHolder = (ConsumerIndexHolder) o; - if (restartTimerOnMaxSize || producerIndex == consumerIndexHolder.index) { + if (!restartTimerOnMaxSize || producerIndex == consumerIndexHolder.index) { w.onComplete(); count = 0; w = UnicastProcessor.create(bufferSize); diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowTimed.java index 58f94304ed..df10e149ba 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowTimed.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowTimed.java @@ -444,7 +444,7 @@ void drainLoop() { if (isHolder) { ConsumerIndexHolder consumerIndexHolder = (ConsumerIndexHolder) o; - if (restartTimerOnMaxSize || producerIndex == consumerIndexHolder.index) { + if (!restartTimerOnMaxSize || producerIndex == consumerIndexHolder.index) { w.onComplete(); count = 0; w = UnicastSubject.create(bufferSize); diff --git a/src/test/java/io/reactivex/rxjava3/flowable/FlowableWindowTests.java b/src/test/java/io/reactivex/rxjava3/flowable/FlowableWindowTests.java index bcf3638316..d43b96f17f 100644 --- a/src/test/java/io/reactivex/rxjava3/flowable/FlowableWindowTests.java +++ b/src/test/java/io/reactivex/rxjava3/flowable/FlowableWindowTests.java @@ -16,11 +16,15 @@ import static org.junit.Assert.*; import java.util.*; +import java.util.concurrent.TimeUnit; import org.junit.Test; import io.reactivex.rxjava3.core.*; import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.processors.PublishProcessor; +import io.reactivex.rxjava3.schedulers.TestScheduler; +import io.reactivex.rxjava3.subscribers.TestSubscriber; public class FlowableWindowTests extends RxJavaTest { @@ -50,4 +54,43 @@ public void accept(List xs) { assertEquals(2, lists.size()); } + + @Test + public void timeSizeWindowAlternatingBounds() { + TestScheduler scheduler = new TestScheduler(); + PublishProcessor pp = PublishProcessor.create(); + + TestSubscriber> ts = pp.window(5, TimeUnit.SECONDS, scheduler, 2) + .flatMapSingle(new Function, SingleSource>>() { + @Override + public SingleSource> apply(Flowable v) throws Throwable { + return v.toList(); + } + }) + .test(); + + pp.onNext(1); + pp.onNext(2); + ts.assertValueCount(1); // size bound hit + + scheduler.advanceTimeBy(1, TimeUnit.SECONDS); + pp.onNext(3); + scheduler.advanceTimeBy(6, TimeUnit.SECONDS); + ts.assertValueCount(2); // time bound hit + + pp.onNext(4); + pp.onNext(5); + + ts.assertValueCount(3); // size bound hit again + + pp.onNext(4); + + scheduler.advanceTimeBy(6, TimeUnit.SECONDS); + + ts.assertValueCount(4) + .assertNoErrors() + .assertNotComplete(); + + ts.cancel(); + } } diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithTimeTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithTimeTest.java index c357800827..7e055bd299 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithTimeTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithTimeTest.java @@ -65,17 +65,19 @@ public void subscribe(Subscriber subscriber) { Flowable> windowed = source.window(100, TimeUnit.MILLISECONDS, scheduler, 2); windowed.subscribe(observeWindow(list, lists)); - scheduler.advanceTimeTo(100, TimeUnit.MILLISECONDS); + scheduler.advanceTimeTo(95, TimeUnit.MILLISECONDS); assertEquals(1, lists.size()); assertEquals(lists.get(0), list("one", "two")); - scheduler.advanceTimeTo(200, TimeUnit.MILLISECONDS); - assertEquals(2, lists.size()); - assertEquals(lists.get(1), list("three", "four")); + scheduler.advanceTimeTo(195, TimeUnit.MILLISECONDS); + assertEquals(3, lists.size()); + assertTrue(lists.get(1).isEmpty()); + assertEquals(lists.get(2), list("three", "four")); scheduler.advanceTimeTo(300, TimeUnit.MILLISECONDS); - assertEquals(3, lists.size()); - assertEquals(lists.get(2), list("five")); + assertEquals(5, lists.size()); + assertTrue(lists.get(3).isEmpty()); + assertEquals(lists.get(4), list("five")); } @Test diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowWithTimeTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowWithTimeTest.java index ca9270448b..e56a094a8f 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowWithTimeTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowWithTimeTest.java @@ -65,17 +65,19 @@ public void subscribe(Observer observer) { Observable> windowed = source.window(100, TimeUnit.MILLISECONDS, scheduler, 2); windowed.subscribe(observeWindow(list, lists)); - scheduler.advanceTimeTo(100, TimeUnit.MILLISECONDS); + scheduler.advanceTimeTo(95, TimeUnit.MILLISECONDS); assertEquals(1, lists.size()); assertEquals(lists.get(0), list("one", "two")); - scheduler.advanceTimeTo(200, TimeUnit.MILLISECONDS); - assertEquals(2, lists.size()); - assertEquals(lists.get(1), list("three", "four")); + scheduler.advanceTimeTo(195, TimeUnit.MILLISECONDS); + assertEquals(3, lists.size()); + assertTrue(lists.get(1).isEmpty()); + assertEquals(lists.get(2), list("three", "four")); scheduler.advanceTimeTo(300, TimeUnit.MILLISECONDS); - assertEquals(3, lists.size()); - assertEquals(lists.get(2), list("five")); + assertEquals(5, lists.size()); + assertTrue(lists.get(3).isEmpty()); + assertEquals(lists.get(4), list("five")); } @Test diff --git a/src/test/java/io/reactivex/rxjava3/observable/ObservableWindowTests.java b/src/test/java/io/reactivex/rxjava3/observable/ObservableWindowTests.java index bfdeed1c6d..1d4a62f7f3 100644 --- a/src/test/java/io/reactivex/rxjava3/observable/ObservableWindowTests.java +++ b/src/test/java/io/reactivex/rxjava3/observable/ObservableWindowTests.java @@ -16,12 +16,16 @@ import static org.junit.Assert.*; import java.util.*; +import java.util.concurrent.TimeUnit; import org.junit.Test; +import io.reactivex.rxjava3.core.*; import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.RxJavaTest; import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.observers.TestObserver; +import io.reactivex.rxjava3.schedulers.*; +import io.reactivex.rxjava3.subjects.PublishSubject; public class ObservableWindowTests extends RxJavaTest { @@ -51,4 +55,43 @@ public void accept(List xs) { assertEquals(2, lists.size()); } + + @Test + public void timeSizeWindowAlternatingBounds() { + TestScheduler scheduler = new TestScheduler(); + PublishSubject ps = PublishSubject.create(); + + TestObserver> to = ps.window(5, TimeUnit.SECONDS, scheduler, 2) + .flatMapSingle(new Function, SingleSource>>() { + @Override + public SingleSource> apply(Observable v) throws Throwable { + return v.toList(); + } + }) + .test(); + + ps.onNext(1); + ps.onNext(2); + to.assertValueCount(1); // size bound hit + + scheduler.advanceTimeBy(1, TimeUnit.SECONDS); + ps.onNext(3); + scheduler.advanceTimeBy(6, TimeUnit.SECONDS); + to.assertValueCount(2); // time bound hit + + ps.onNext(4); + ps.onNext(5); + + to.assertValueCount(3); // size bound hit again + + ps.onNext(4); + + scheduler.advanceTimeBy(6, TimeUnit.SECONDS); + + to.assertValueCount(4) + .assertNoErrors() + .assertNotComplete(); + + to.dispose(); + } }