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

* 2.x: improve Flowable.timeout()

* Remove the now unused FullArbiter(Subscriber)

* Don't read the volatile twice

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.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) {

src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.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;