Merge branch 'OperatorDefer' of github.com:akarnokd/RxJava into merge-prs

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

Author: benjchristensen

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

@@ -49,7 +49,6 @@
 import rx.operators.OnSubscribeFromIterable;
 import rx.operators.OnSubscribeRange;
 import rx.operators.OperationDebounce;
-import rx.operators.OperationDefer;
 import rx.operators.OperationDelay;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
@@ -98,6 +97,7 @@
 import rx.operators.OperatorCombineLatest;
 import rx.operators.OperatorConcat;
 import rx.operators.OperatorDefaultIfEmpty;
+import rx.operators.OperatorDefer;
 import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorElementAt;
 import rx.operators.OperatorFilter;
@@ -1007,7 +1007,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#wiki-defer">RxJava Wiki: defer()</a>
      */
     public final static <T> Observable<T> defer(Func0<? extends Observable<? extends T>> observableFactory) {
-        return create(OperationDefer.defer(observableFactory));
+        return create(new OperatorDefer<T>(observableFactory));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationDefer.java renamed to rxjava-core/src/main/java/rx/operators/OperatorDefer.java

@@ -16,11 +16,9 @@
 package rx.operators;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
-import rx.Subscription;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
 import rx.functions.Func0;
-import rx.observers.Subscribers;
 
 /**
  * Do not create the Observable until an Observer subscribes; create a fresh Observable on each
@@ -32,17 +30,23 @@
  * return an Observable that will call this function to generate its Observable sequence afresh
  * each time a new Observer subscribes.
  */
-public final class OperationDefer {
+public final class OperatorDefer<T> implements OnSubscribe<T> {
+    final Func0<? extends Observable<? extends T>> observableFactory;
 
-    public static <T> OnSubscribeFunc<T> defer(final Func0<? extends Observable<? extends T>> observableFactory) {
-
-        return new OnSubscribeFunc<T>() {
-            @Override
-            public Subscription onSubscribe(Observer<? super T> observer) {
-                Observable<? extends T> obs = observableFactory.call();
-                return obs.unsafeSubscribe(Subscribers.from(observer));
-            }
-        };
+    public OperatorDefer(Func0<? extends Observable<? extends T>> observableFactory) {
+        this.observableFactory = observableFactory;
+    }
 
+    @Override
+    public void call(Subscriber<? super T> s) {
+        Observable<? extends T> o;
+        try {
+            o = observableFactory.call();
+        } catch (Throwable t) {
+            s.onError(t);
+            return;
+        }
+        o.unsafeSubscribe(s);
     }
+    
 }

rxjava-core/src/test/java/rx/operators/OperationDeferTest.java renamed to rxjava-core/src/test/java/rx/operators/OperatorDeferTest.java

@@ -15,22 +15,18 @@
  */
 package rx.operators;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyZeroInteractions;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Observer;
 import rx.functions.Func0;
 
-public class OperationDeferTest {
+@SuppressWarnings("unchecked")
+public class OperatorDeferTest {
 
     @Test
-    @SuppressWarnings("unchecked")
     public void testDefer() throws Throwable {
 
         Func0<Observable<String>> factory = mock(Func0.class);
@@ -64,4 +60,21 @@ public void testDefer() throws Throwable {
         verify(secondObserver, times(1)).onCompleted();
 
     }
+    
+    @Test
+    public void testDeferFunctionThrows() {
+        Func0<Observable<String>> factory = mock(Func0.class);
+        
+        when(factory.call()).thenThrow(new OperationReduceTest.CustomException());
+        
+        Observable<String> result = Observable.defer(factory);
+        
+        Observer<String> o = mock(Observer.class);
+        
+        result.subscribe(o);
+        
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+        verify(o, never()).onNext(any(String.class));
+        verify(o, never()).onCompleted();
+    }
 }