Fix BlockingOperator Subscribe Behavior

The blocking operators need to use 'subscribe', not 'unsafeSubscribe' since they need the SafeSubscriber behavior in the final subscribe step they perform.
Renamed operators to BlockingOperator* for clarity and to match the Operation->Operator naming change.

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -74,7 +74,7 @@
 import rx.operators.OperationInterval;
 import rx.operators.OperationJoin;
 import rx.operators.OperationJoinPatterns;
-import rx.operators.OperationMaterialize;
+import rx.operators.OperatorMaterialize;
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMergeMaxConcurrent;
 import rx.operators.OperationMinMax;
@@ -4915,7 +4915,7 @@ public final <R> Observable<R> mapMany(Func1<? super T, ? extends Observable<R>>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229453.aspx">MSDN: Observable.materialize</a>
      */
     public final Observable<Notification<T>> materialize() {
-        return create(OperationMaterialize.materialize(this));
+        return lift(new OperatorMaterialize<T>());
     }
 
     /**

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -22,15 +22,13 @@
 
 import rx.Observable;
 import rx.Subscriber;
-import rx.Subscription;
 import rx.functions.Action1;
 import rx.functions.Func1;
-import rx.observers.SafeSubscriber;
-import rx.operators.OperationLatest;
-import rx.operators.OperationMostRecent;
-import rx.operators.OperationNext;
-import rx.operators.OperationToFuture;
-import rx.operators.OperationToIterator;
+import rx.operators.BlockingOperatorLatest;
+import rx.operators.BlockingOperatorMostRecent;
+import rx.operators.BlockingOperatorNext;
+import rx.operators.BlockingOperatorToFuture;
+import rx.operators.BlockingOperatorToIterator;
 
 /**
  * An extension of {@link Observable} that provides blocking operators.
@@ -64,17 +62,6 @@ public static <T> BlockingObservable<T> from(final Observable<? extends T> o) {
         return new BlockingObservable<T>(o);
     }
 
-    /**
-     * Used for protecting against errors being thrown from {@link Subscriber} implementations and ensuring onNext/onError/onCompleted contract
-     * compliance.
-     * <p>
-     * See https://github.com/Netflix/RxJava/issues/216 for discussion on
-     * "Guideline 6.4: Protect calls to user code from within an operator"
-     */
-    private Subscription protectivelyWrapAndSubscribe(Subscriber<? super T> observer) {
-        return o.unsafeSubscribe(new SafeSubscriber<T>(observer));
-    }
-
     /**
      * Invoke a method on each item emitted by the {@link Observable}; block
      * until the Observable completes.
@@ -97,12 +84,10 @@ public void forEach(final Action1<? super T> onNext) {
         final AtomicReference<Throwable> exceptionFromOnError = new AtomicReference<Throwable>();
 
         /**
-         * Wrapping since raw functions provided by the user are being invoked.
-         * 
-         * See https://github.com/Netflix/RxJava/issues/216 for discussion on
-         * "Guideline 6.4: Protect calls to user code from within an operator"
+         * Use 'subscribe' instead of 'unsafeSubscribe' for Rx contract behavior
+         * as this is the final subscribe in the chain.
          */
-        protectivelyWrapAndSubscribe(new Subscriber<T>() {
+        o.subscribe(new Subscriber<T>() {
             @Override
             public void onCompleted() {
                 latch.countDown();
@@ -158,7 +143,7 @@ public void onNext(T args) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Blocking-Observable-Operators#transformations-tofuture-toiterable-and-toiteratorgetiterator">RxJava Wiki: getIterator()</a>
      */
     public Iterator<T> getIterator() {
-        return OperationToIterator.toIterator(o);
+        return BlockingOperatorToIterator.toIterator(o);
     }
 
     /**
@@ -311,7 +296,7 @@ public T lastOrDefault(T defaultValue, Func1<? super T, Boolean> predicate) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229751.aspx">MSDN: Observable.MostRecent</a>
      */
     public Iterable<T> mostRecent(T initialValue) {
-        return OperationMostRecent.mostRecent(o, initialValue);
+        return BlockingOperatorMostRecent.mostRecent(o, initialValue);
     }
 
     /**
@@ -324,7 +309,7 @@ public Iterable<T> mostRecent(T initialValue) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211897.aspx">MSDN: Observable.Next</a>
      */
     public Iterable<T> next() {
-        return OperationNext.next(o);
+        return BlockingOperatorNext.next(o);
     }
 
     /**
@@ -344,7 +329,7 @@ public Iterable<T> next() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212115.aspx">MSDN: Observable.Latest</a>
      */
     public Iterable<T> latest() {
-        return OperationLatest.latest(o);
+        return BlockingOperatorLatest.latest(o);
     }
 
     /**
@@ -441,7 +426,7 @@ public T singleOrDefault(T defaultValue, Func1<? super T, Boolean> predicate) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Blocking-Observable-Operators#transformations-tofuture-toiterable-and-toiteratorgetiterator">RxJava Wiki: toFuture()</a>
      */
     public Future<T> toFuture() {
-        return OperationToFuture.toFuture(o);
+        return BlockingOperatorToFuture.toFuture(o);
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationLatest.java renamed to rxjava-core/src/main/java/rx/operators/BlockingOperatorLatest.java

@@ -30,9 +30,9 @@
  * If the source works faster than the iterator, values may be skipped, but
  * not the onError or onCompleted events.
  */
-public final class OperationLatest {
+public final class BlockingOperatorLatest {
     /** Utility class. */
-    private OperationLatest() {
+    private BlockingOperatorLatest() {
         throw new IllegalStateException("No instances!");
     }
 
@@ -41,7 +41,7 @@ public static <T> Iterable<T> latest(final Observable<? extends T> source) {
             @Override
             public Iterator<T> iterator() {
                 LatestObserverIterator<T> lio = new LatestObserverIterator<T>();
-                source.materialize().unsafeSubscribe(lio);
+                source.materialize().subscribe(lio);
                 return lio;
             }
         };

rxjava-core/src/main/java/rx/operators/OperationMostRecent.java renamed to rxjava-core/src/main/java/rx/operators/BlockingOperatorMostRecent.java

@@ -29,7 +29,7 @@
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.mostRecent.png">
  */
-public final class OperationMostRecent {
+public final class BlockingOperatorMostRecent {
 
     public static <T> Iterable<T> mostRecent(final Observable<? extends T> source, final T initialValue) {
 
@@ -39,7 +39,11 @@ public Iterator<T> iterator() {
                 MostRecentObserver<T> mostRecentObserver = new MostRecentObserver<T>(initialValue);
                 final MostRecentIterator<T> nextIterator = new MostRecentIterator<T>(mostRecentObserver);
 
-                source.unsafeSubscribe(mostRecentObserver);
+                /**
+                 * Subscribe instead of unsafeSubscribe since this is the final subscribe in the chain
+                 * since it is for BlockingObservable.
+                 */
+                source.subscribe(mostRecentObserver);
 
                 return nextIterator;
             }

rxjava-core/src/main/java/rx/operators/OperationNext.java renamed to rxjava-core/src/main/java/rx/operators/BlockingOperatorNext.java

@@ -31,7 +31,7 @@
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.next.png">
  */
-public final class OperationNext {
+public final class BlockingOperatorNext {
 
     public static <T> Iterable<T> next(final Observable<? extends T> items) {
         return new Iterable<T>() {
@@ -40,7 +40,7 @@ public Iterator<T> iterator() {
                 NextObserver<T> nextObserver = new NextObserver<T>();
                 final NextIterator<T> nextIterator = new NextIterator<T>(nextObserver);
 
-                items.materialize().unsafeSubscribe(nextObserver);
+                items.materialize().subscribe(nextObserver);
 
                 return nextIterator;
             }

rxjava-core/src/main/java/rx/operators/OperationToFuture.java renamed to rxjava-core/src/main/java/rx/operators/BlockingOperatorToFuture.java

@@ -35,7 +35,7 @@
  * The toFuture operation throws an exception if the Observable emits more than one item. If the
  * Observable may emit more than item, use <code>toList().toFuture()</code>.
  */
-public class OperationToFuture {
+public class BlockingOperatorToFuture {
 
     /**
      * Returns a Future that expects a single item from the observable.
@@ -52,7 +52,7 @@ public static <T> Future<T> toFuture(Observable<? extends T> that) {
         final AtomicReference<T> value = new AtomicReference<T>();
         final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
 
-        final Subscription s = that.unsafeSubscribe(new Subscriber<T>() {
+        final Subscription s = that.subscribe(new Subscriber<T>() {
 
             @Override
             public void onCompleted() {

rxjava-core/src/main/java/rx/operators/OperationToIterator.java renamed to rxjava-core/src/main/java/rx/operators/BlockingOperatorToIterator.java

@@ -33,7 +33,7 @@
  * 
  * @see <a href="https://github.com/Netflix/RxJava/issues/50">Issue #50</a>
  */
-public class OperationToIterator {
+public class BlockingOperatorToIterator {
 
     /**
      * Returns an iterator that iterates all values of the observable.
@@ -45,7 +45,8 @@ public class OperationToIterator {
     public static <T> Iterator<T> toIterator(Observable<? extends T> source) {
         final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>();
 
-        source.materialize().unsafeSubscribe(new Subscriber<Notification<? extends T>>() {
+        // using subscribe instead of unsafeSubscribe since this is a BlockingObservable "final subscribe"
+        source.materialize().subscribe(new Subscriber<Notification<? extends T>>() {
             @Override
             public void onCompleted() {
                 // ignore

rxjava-core/src/main/java/rx/operators/OperationDematerialize.java

@@ -23,7 +23,7 @@
 import rx.Subscription;
 
 /**
- * Reverses the effect of {@link OperationMaterialize} by transforming the Notification objects
+ * Reverses the effect of {@link OperatorMaterialize} by transforming the Notification objects
  * emitted by a source Observable into the items or notifications they represent.
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/dematerialize.png">

rxjava-core/src/main/java/rx/operators/OperationMaterialize.java

@@ -0,0 +1,56 @@
+/**
+ * Copyright 2014 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 rx.operators;
+
+import rx.Notification;
+import rx.Observable.Operator;
+import rx.Subscriber;
+
+/**
+ * Turns all of the notifications from an Observable into <code>onNext</code> emissions, and marks
+ * them with their original notification types within {@link Notification} objects.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/materialize.png">
+ * <p>
+ * See <a href="http://msdn.microsoft.com/en-us/library/hh229453(v=VS.103).aspx">here</a> for the
+ * Microsoft Rx equivalent.
+ */
+public final class OperatorMaterialize<T> implements Operator<Notification<T>, T> {
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super Notification<T>> child) {
+        return new Subscriber<T>(child) {
+
+            @Override
+            public void onCompleted() {
+                child.onNext(Notification.<T> createOnCompleted());
+                child.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onNext(Notification.<T> createOnError(e));
+                child.onCompleted();
+            }
+
+            @Override
+            public void onNext(T t) {
+                child.onNext(Notification.<T> createOnNext(t));
+            }
+
+        };
+    }
+}