2.x: improve Flowable.timeout() (#5661)

* 2.x: improve Flowable.timeout() * Remove the now unused FullArbiter(Subscriber) * Don't read the volatile twice
ReactiveX · Oct 13, 2017 · 5387e20 · 5387e20
1 parent bffc7f2
commit 5387e20
Show file tree

Hide file tree

Showing 9 changed files with 692 additions and 748 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java
diff --git a/src/main/java/io/reactivex/internal/subscribers/FullArbiterSubscriber.java b/src/main/java/io/reactivex/internal/subscribers/FullArbiterSubscriber.java
diff --git a/src/main/java/io/reactivex/internal/subscriptions/FullArbiter.java b/src/main/java/io/reactivex/internal/subscriptions/FullArbiter.java
diff --git a/src/main/java/io/reactivex/schedulers/TestScheduler.java b/src/main/java/io/reactivex/schedulers/TestScheduler.java
@@ -102,14 +102,14 @@ public void triggerActions() {
     }
 
     private void triggerActions(long targetTimeInNanoseconds) {
-        while (!queue.isEmpty()) {
+        for (;;) {
             TimedRunnable current = queue.peek();
-            if (current.time > targetTimeInNanoseconds) {
+            if (current == null || current.time > targetTimeInNanoseconds) {
                 break;
             }
             // if scheduled time is 0 (immediate) use current virtual time
             time = current.time == 0 ? time : current.time;
-            queue.remove();
+            queue.remove(current);
 
             // Only execute if not unsubscribed
             if (!current.scheduler.disposed) {

diff --git a/src/test/java/io/reactivex/flowable/FlowableNullTests.java b/src/test/java/io/reactivex/flowable/FlowableNullTests.java
@@ -2328,7 +2328,7 @@ public void timeoutFirstItemNull() {
 
     @Test(expected = NullPointerException.class)
     public void timeoutFirstItemReturnsNull() {
-        just1.timeout(just1, new Function<Integer, Publisher<Object>>() {
+        just1.timeout(Flowable.never(), new Function<Integer, Publisher<Object>>() {
             @Override
             public Publisher<Object> apply(Integer v) {
                 return null;