2.x: reduce blockingX overhead, move internal observers to common pkg

src/main/java/io/reactivex/Completable.java

@@ -21,12 +21,12 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
+import io.reactivex.internal.observers.*;
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.operators.flowable.FlowableDelaySubscriptionOther;
 import io.reactivex.internal.operators.maybe.MaybeFromCompletable;
 import io.reactivex.internal.operators.observable.ObservableDelaySubscriptionOther;
 import io.reactivex.internal.operators.single.SingleDelayWithCompletable;
-import io.reactivex.internal.subscribers.completable.*;
 import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
@@ -795,7 +795,9 @@ public final Completable andThen(CompletableSource next) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final void blockingAwait() {
-        CompletableAwait.await(this);
+        BlockingObserver<Void> observer = new BlockingObserver<Void>();
+        subscribe(observer);
+        observer.blockingGet();
     }
 
     /**
@@ -813,7 +815,9 @@ public final void blockingAwait() {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final boolean blockingAwait(long timeout, TimeUnit unit) {
-        return CompletableAwait.await(this, timeout, unit);
+        BlockingObserver<Void> observer = new BlockingObserver<Void>();
+        subscribe(observer);
+        return observer.blockingAwait(timeout, unit);
     }
 
     /**
@@ -828,7 +832,9 @@ public final boolean blockingAwait(long timeout, TimeUnit unit) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Throwable blockingGet() {
-        return CompletableAwait.get(this);
+        BlockingObserver<Void> observer = new BlockingObserver<Void>();
+        subscribe(observer);
+        return observer.blockingGetError();
     }
 
     /**
@@ -842,7 +848,10 @@ public final Throwable blockingGet() {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Throwable blockingGet(long timeout, TimeUnit unit) {
-        return CompletableAwait.get(this, timeout, unit);
+        ObjectHelper.requireNonNull(unit, "unit is null");
+        BlockingObserver<Void> observer = new BlockingObserver<Void>();
+        subscribe(observer);
+        return observer.blockingGetError(timeout, unit);
     }
 
     /**

src/main/java/io/reactivex/Maybe.java

@@ -22,6 +22,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
+import io.reactivex.internal.observers.BlockingObserver;
 import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.operators.maybe.*;
 import io.reactivex.internal.util.*;
@@ -710,7 +711,9 @@ public static <T> Maybe<T> wrap(MaybeSource<T> source) {
      * @return the success value
      */
     public T blockingGet() {
-        return MaybeAwait.get(this, null);
+        BlockingObserver<T> observer = new BlockingObserver<T>();
+        subscribe(observer);
+        return observer.blockingGet();
     }
 
     /**
@@ -725,7 +728,9 @@ public T blockingGet() {
      */
     public T blockingGet(T defaultValue) {
         ObjectHelper.requireNonNull(defaultValue, "defaultValue is null");
-        return MaybeAwait.get(this, defaultValue);
+        BlockingObserver<T> observer = new BlockingObserver<T>();
+        subscribe(observer);
+        return observer.blockingGet(defaultValue);
     }
 
     /**

src/main/java/io/reactivex/Single.java

@@ -23,12 +23,12 @@
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
 import io.reactivex.internal.fuseable.FuseToFlowable;
+import io.reactivex.internal.observers.*;
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.operators.maybe.*;
 import io.reactivex.internal.operators.observable.ObservableConcatMap;
 import io.reactivex.internal.operators.single.*;
-import io.reactivex.internal.subscribers.single.*;
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
@@ -1775,7 +1775,9 @@ public final Completable flatMapCompletable(final Function<? super T, ? extends
      * @return the success value
      */
     public final T blockingGet() {
-        return SingleAwait.get(this);
+        BlockingObserver<T> observer = new BlockingObserver<T>();
+        subscribe(observer);
+        return observer.blockingGet();
     }
 
     /**

src/main/java/io/reactivex/internal/subscribers/single/BiConsumerSingleObserver.java → src/main/java/io/reactivex/internal/observers/BiConsumerSingleObserver.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.single;
+package io.reactivex.internal.observers;
 
 import java.util.concurrent.atomic.AtomicReference;
 

src/main/java/io/reactivex/internal/observers/BlockingObserver.java

@@ -0,0 +1,209 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.observers;
+
+import java.util.concurrent.*;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.util.ExceptionHelper;
+
+/**
+ * A combined Observer that awaits the success or error signal via a CountDownLatch.
+ * @param <T> the value type
+ */
+public final class BlockingObserver<T>
+extends CountDownLatch
+implements SingleObserver<T>, CompletableObserver, MaybeObserver<T> {
+
+    T value;
+    Throwable error;
+
+    Disposable d;
+
+    volatile boolean cancelled;
+
+    public BlockingObserver() {
+        super(1);
+    }
+
+    void dispose() {
+        cancelled = true;
+        Disposable d = this.d;
+        if (d != null) {
+            d.dispose();
+        }
+    }
+
+    @Override
+    public void onSubscribe(Disposable d) {
+        this.d = d;
+        if (cancelled) {
+            d.dispose();
+        }
+    }
+
+    @Override
+    public void onSuccess(T value) {
+        this.value = value;
+        countDown();
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        error = e;
+        countDown();
+    }
+
+    @Override
+    public void onComplete() {
+        countDown();
+    }
+
+    /**
+     * Block until the latch is counted down then rethrow any exception received (wrapped if checked)
+     * or return the received value (null if none).
+     * @return the value received or null if no value received
+     */
+    public T blockingGet() {
+        if (getCount() != 0) {
+            try {
+                await();
+            } catch (InterruptedException ex) {
+                dispose();
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+        Throwable ex = error;
+        if (ex != null) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+        return value;
+    }
+
+    /**
+     * Block until a the latch is counted down and return the value received, otherwise
+     * rethrow the received exception or rethrow the InterruptedException or TimeoutException
+     * (wrapped).
+     * @param timeout the timeout value
+     * @param unit the time unit
+     * @return the value received or null if no value received
+     */
+    public T blockingGet(long timeout, TimeUnit unit) {
+        if (getCount() != 0) {
+            try {
+                if (!await(timeout, unit)) {
+                    dispose();
+                    throw ExceptionHelper.wrapOrThrow(new TimeoutException());
+                }
+            } catch (InterruptedException ex) {
+                dispose();
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+        Throwable ex = error;
+        if (ex != null) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+        return value;
+    }
+
+    /**
+     * Block until the latch is counted down then rethrow any exception received (wrapped if checked)
+     * or return the received value (the defaultValue if none).
+     * @param defaultValue the default value to return if no value was received
+     * @return the value received or defaultValue if no value received
+     */
+    public T blockingGet(T defaultValue) {
+        if (getCount() != 0) {
+            try {
+                await();
+            } catch (InterruptedException ex) {
+                dispose();
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+        Throwable ex = error;
+        if (ex != null) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+        T v = value;
+        return v != null ? v : defaultValue;
+    }
+
+    /**
+     * Block until the latch is counted down and return the error received or null if no
+     * error happened.
+     * @return the error received or null
+     */
+    public Throwable blockingGetError() {
+        if (getCount() != 0) {
+            try {
+                await();
+            } catch (InterruptedException ex) {
+                dispose();
+                return ex;
+            }
+        }
+        return error;
+    }
+
+    /**
+     * Block until the latch is counted down and return the error received or
+     * when the wait is interrupted or times out, null otherwise.
+     * @param timeout the timeout value
+     * @param unit the time unit
+     * @return the error received or null
+     */
+    public Throwable blockingGetError(long timeout, TimeUnit unit) {
+        if (getCount() != 0) {
+            try {
+                if (!await(timeout, unit)) {
+                    dispose();
+                    throw ExceptionHelper.wrapOrThrow(new TimeoutException());
+                }
+            } catch (InterruptedException ex) {
+                dispose();
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+        return error;
+    }
+
+    /**
+     * Block until the observer terminates and return true; return false if 
+     * the wait times out.
+     * @param timeout the timeout value
+     * @param unit the time unit
+     * @return true if the observer terminated in time, false otherwise
+     */
+    public boolean blockingAwait(long timeout, TimeUnit unit) {
+        if (getCount() != 0) {
+            try {
+                if (!await(timeout, unit)) {
+                    dispose();
+                    return false;
+                }
+            } catch (InterruptedException ex) {
+                dispose();
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+        Throwable ex = error;
+        if (ex != null) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+        return true;
+    }
+}

src/main/java/io/reactivex/internal/subscribers/completable/CallbackCompletableObserver.java → src/main/java/io/reactivex/internal/observers/CallbackCompletableObserver.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.completable;
+package io.reactivex.internal.observers;
 
 import java.util.concurrent.atomic.AtomicReference;
 

src/main/java/io/reactivex/internal/subscribers/single/ConsumerSingleObserver.java → src/main/java/io/reactivex/internal/observers/ConsumerSingleObserver.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.single;
+package io.reactivex.internal.observers;
 
 import java.util.concurrent.atomic.AtomicReference;
 

src/main/java/io/reactivex/internal/subscribers/completable/EmptyCompletableObserver.java → src/main/java/io/reactivex/internal/observers/EmptyCompletableObserver.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.completable;
+package io.reactivex.internal.observers;
 
 import java.util.concurrent.atomic.AtomicReference;
 

src/main/java/io/reactivex/internal/subscribers/single/FutureSingleObserver.java → src/main/java/io/reactivex/internal/observers/FutureSingleObserver.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.single;
+package io.reactivex.internal.observers;
 
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;

src/main/java/io/reactivex/internal/subscribers/completable/SubscriberCompletableObserver.java → src/main/java/io/reactivex/internal/observers/SubscriberCompletableObserver.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.completable;
+package io.reactivex.internal.observers;
 
 import org.reactivestreams.*;