2.x: cleanup, fixes, coverage 10/24-1

src/main/java/io/reactivex/Flowable.java

@@ -12019,7 +12019,7 @@ public final Flowable<T> startWithArray(T... items) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe() {
         return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER,
-                Functions.EMPTY_ACTION, FlowableInternalHelper.requestMax());
+                Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);
     }
 
     /**
@@ -12046,7 +12046,7 @@ public final Disposable subscribe() {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe(Consumer<? super T> onNext) {
         return subscribe(onNext, Functions.ERROR_CONSUMER,
-                Functions.EMPTY_ACTION, FlowableInternalHelper.requestMax());
+                Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);
     }
 
     /**
@@ -12075,7 +12075,7 @@ public final Disposable subscribe(Consumer<? super T> onNext) {
     @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
-        return subscribe(onNext, onError, Functions.EMPTY_ACTION, FlowableInternalHelper.requestMax());
+        return subscribe(onNext, onError, Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);
     }
 
     /**
@@ -12109,7 +12109,7 @@ public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super T
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
             Action onComplete) {
-        return subscribe(onNext, onError, onComplete, FlowableInternalHelper.requestMax());
+        return subscribe(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
     }
 
     /**

src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java

@@ -13,9 +13,17 @@
 
 package io.reactivex.internal.operators.flowable;
 
-import java.util.Iterator;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.*;
 
-import org.reactivestreams.Publisher;
+import org.reactivestreams.*;
+
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.MissingBackpressureException;
+import io.reactivex.internal.queue.SpscArrayQueue;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.internal.util.ExceptionHelper;
 
 public final class BlockingFlowableIterable<T> implements Iterable<T> {
     final Publisher<? extends T> source;
@@ -33,4 +41,145 @@ public Iterator<T> iterator() {
         source.subscribe(it);
         return it;
     }
+
+    static final class BlockingFlowableIterator<T>
+    extends AtomicReference<Subscription>
+    implements Subscriber<T>, Iterator<T>, Runnable, Disposable {
+
+        private static final long serialVersionUID = 6695226475494099826L;
+
+        final SpscArrayQueue<T> queue;
+
+        final long batchSize;
+
+        final long limit;
+
+        final Lock lock;
+
+        final Condition condition;
+
+        long produced;
+
+        volatile boolean done;
+        Throwable error;
+
+        BlockingFlowableIterator(int batchSize) {
+            this.queue = new SpscArrayQueue<T>(batchSize);
+            this.batchSize = batchSize;
+            this.limit = batchSize - (batchSize >> 2);
+            this.lock = new ReentrantLock();
+            this.condition = lock.newCondition();
+        }
+
+        @Override
+        public boolean hasNext() {
+            for (;;) {
+                boolean d = done;
+                boolean empty = queue.isEmpty();
+                if (d) {
+                    Throwable e = error;
+                    if (e != null) {
+                        throw ExceptionHelper.wrapOrThrow(e);
+                    } else
+                    if (empty) {
+                        return false;
+                    }
+                }
+                if (empty) {
+                    lock.lock();
+                    try {
+                        while (!done && queue.isEmpty()) {
+                            condition.await();
+                        }
+                    } catch (InterruptedException ex) {
+                        run();
+                        throw ExceptionHelper.wrapOrThrow(ex);
+                    } finally {
+                        lock.unlock();
+                    }
+                } else {
+                    return true;
+                }
+            }
+        }
+
+        @Override
+        public T next() {
+            if (hasNext()) {
+                T v = queue.poll();
+
+                long p = produced + 1;
+                if (p == limit) {
+                    produced = 0;
+                    get().request(p);
+                } else {
+                    produced = p;
+                }
+
+                return v;
+            }
+            throw new NoSuchElementException();
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (SubscriptionHelper.setOnce(this, s)) {
+                s.request(batchSize);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            if (!queue.offer(t)) {
+                SubscriptionHelper.cancel(this);
+
+                onError(new MissingBackpressureException("Queue full?!"));
+            } else {
+                signalConsumer();
+            }
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            error = t;
+            done = true;
+            signalConsumer();
+        }
+
+        @Override
+        public void onComplete() {
+            done = true;
+            signalConsumer();
+        }
+
+        void signalConsumer() {
+            lock.lock();
+            try {
+                condition.signalAll();
+            } finally {
+                lock.unlock();
+            }
+        }
+
+        @Override
+        public void run() {
+            SubscriptionHelper.cancel(this);
+            signalConsumer();
+        }
+
+        @Override // otherwise default method which isn't available in Java 7
+        public void remove() {
+            throw new UnsupportedOperationException("remove");
+        }
+
+        @Override
+        public void dispose() {
+            SubscriptionHelper.cancel(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return SubscriptionHelper.isCancelled(get());
+        }
+    }
 }