2.x: coverage, minor fixes, cleanup 10/19-2 (#4732)

Author: akarnokd

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

@@ -12,12 +12,13 @@
  */
 package io.reactivex.internal.operators.flowable;
 
+import org.reactivestreams.*;
+
 import io.reactivex.Flowable;
+import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
-import io.reactivex.internal.subscribers.SubscriptionLambdaSubscriber;
-
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
+import io.reactivex.internal.subscriptions.*;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class FlowableDoOnLifecycle<T> extends AbstractFlowableWithUpstream<T, T> {
     private final Consumer<? super Subscription> onSubscribe;
@@ -36,4 +37,78 @@ public FlowableDoOnLifecycle(Flowable<T> source, Consumer<? super Subscription>
     protected void subscribeActual(Subscriber<? super T> s) {
         source.subscribe(new SubscriptionLambdaSubscriber<T>(s, onSubscribe, onRequest, onCancel));
     }
+
+    static final class SubscriptionLambdaSubscriber<T> implements Subscriber<T>, Subscription {
+        final Subscriber<? super T> actual;
+        final Consumer<? super Subscription> onSubscribe;
+        final LongConsumer onRequest;
+        final Action onCancel;
+
+        Subscription s;
+
+        public SubscriptionLambdaSubscriber(Subscriber<? super T> actual,
+                Consumer<? super Subscription> onSubscribe,
+                LongConsumer onRequest,
+                Action onCancel) {
+            this.actual = actual;
+            this.onSubscribe = onSubscribe;
+            this.onCancel = onCancel;
+            this.onRequest = onRequest;
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            // this way, multiple calls to onSubscribe can show up in tests that use doOnSubscribe to validate behavior
+            try {
+                onSubscribe.accept(s);
+            } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
+                s.cancel();
+                RxJavaPlugins.onError(e);
+                EmptySubscription.error(e, actual);
+                return;
+            }
+            if (SubscriptionHelper.validate(this.s, s)) {
+                this.s = s;
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            actual.onNext(t);
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            actual.onError(t);
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+        }
+
+        @Override
+        public void request(long n) {
+            try {
+                onRequest.accept(n);
+            } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
+                RxJavaPlugins.onError(e);
+            }
+            s.request(n);
+        }
+
+        @Override
+        public void cancel() {
+            try {
+                onCancel.run();
+            } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
+                RxJavaPlugins.onError(e);
+            }
+            s.cancel();
+        }
+    }
 }

src/main/java/io/reactivex/internal/operators/observable/BlockingObservableIterable.java

@@ -13,9 +13,15 @@
 
 package io.reactivex.internal.operators.observable;
 
-import java.util.Iterator;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.*;
 
 import io.reactivex.ObservableSource;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.queue.SpscLinkedArrayQueue;
+import io.reactivex.internal.util.ExceptionHelper;
 
 public final class BlockingObservableIterable<T> implements Iterable<T> {
     final ObservableSource<? extends T> source;
@@ -33,4 +39,118 @@ public Iterator<T> iterator() {
         source.subscribe(it);
         return it;
     }
+    
+    static final class BlockingObservableIterator<T>
+    extends AtomicReference<Disposable>
+    implements io.reactivex.Observer<T>, Iterator<T>, Disposable {
+
+
+        private static final long serialVersionUID = 6695226475494099826L;
+
+        final SpscLinkedArrayQueue<T> queue;
+
+        final Lock lock;
+
+        final Condition condition;
+
+        volatile boolean done;
+        Throwable error;
+
+        public BlockingObservableIterator(int batchSize) {
+            this.queue = new SpscLinkedArrayQueue<T>(batchSize);
+            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) {
+                    try {
+                        lock.lock();
+                        try {
+                            while (!done && queue.isEmpty()) {
+                                condition.await();
+                            }
+                        } finally {
+                            lock.unlock();
+                        }
+                    } catch (InterruptedException ex) {
+                        DisposableHelper.dispose(this);
+                        signalConsumer();
+                        throw ExceptionHelper.wrapOrThrow(ex);
+                    }
+                } else {
+                    return true;
+                }
+            }
+        }
+
+        @Override
+        public T next() {
+            if (hasNext()) {
+                return queue.poll();
+            }
+            throw new NoSuchElementException();
+        }
+
+        @Override
+        public void onSubscribe(Disposable s) {
+            DisposableHelper.setOnce(this, s);
+        }
+
+        @Override
+        public void onNext(T t) {
+            queue.offer(t);
+            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 // otherwise default method which isn't available in Java 7
+        public void remove() {
+            throw new UnsupportedOperationException("remove");
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+    }
 }

src/main/java/io/reactivex/internal/operators/observable/BlockingObservableIterator.java

@@ -121,6 +121,7 @@ public void onError(Throwable e) {
                 RxJavaPlugins.onError(e);
                 return;
             }
+            d = DisposableHelper.DISPOSED;
             actual.onError(e);
         }
 
@@ -129,6 +130,7 @@ public void onComplete() {
             if (d == DisposableHelper.DISPOSED) {
                 return;
             }
+            d = DisposableHelper.DISPOSED;
             actual.onComplete();
         }