@Test public void testUsingWithObservableFactoryErrorInOnSubscribe() { final Action0 unsubscribe = mock(Action0.class); Func0<Subscription> resourceFactory = new Func0<Subscription>() { @Override public Subscription call() { return Subscriptions.create(unsubscribe); } }; Func1<Subscription, Observable<Integer>> observableFactory = new Func1<Subscription, Observable<Integer>>() { @Override public Observable<Integer> call(Subscription subscription) { return Observable.create(new OnSubscribeFunc<Integer>() { @Override public Subscription onSubscribe(Observer<? super Integer> t1) { throw new TestException(); } }); } }; try { Observable.create(using(resourceFactory, observableFactory)) .toBlockingObservable().last(); fail("Should throw a TestException when the observableFactory throws it"); } catch (TestException e) { verify(unsubscribe, times(1)).call(); } }
public static <T, RESOURCE extends Subscription> OnSubscribeFunc<T> using( final Func0<RESOURCE> resourceFactory, final Func1<RESOURCE, Observable<T>> observableFactory) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { Subscription resourceSubscription = Subscriptions.empty(); try { RESOURCE resource = resourceFactory.call(); if (resource != null) { resourceSubscription = resource; } Observable<T> observable = observableFactory.call(resource); SafeObservableSubscription subscription = new SafeObservableSubscription(); return subscription.wrap(new CompositeSubscription( observable.subscribe(new SafeObserver<T>( subscription, observer)), resourceSubscription)); } catch (Throwable e) { resourceSubscription.unsubscribe(); return Observable.<T> error(e).subscribe(observer); } } }; }
OperationUsing { public static <T, RESOURCE extends Subscription> OnSubscribeFunc<T> using( final Func0<RESOURCE> resourceFactory, final Func1<RESOURCE, Observable<T>> observableFactory) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { Subscription resourceSubscription = Subscriptions.empty(); try { RESOURCE resource = resourceFactory.call(); if (resource != null) { resourceSubscription = resource; } Observable<T> observable = observableFactory.call(resource); SafeObservableSubscription subscription = new SafeObservableSubscription(); return subscription.wrap(new CompositeSubscription( observable.subscribe(new SafeObserver<T>( subscription, observer)), resourceSubscription)); } catch (Throwable e) { resourceSubscription.unsubscribe(); return Observable.<T> error(e).subscribe(observer); } } }; } }
OperationUsing { public static <T, RESOURCE extends Subscription> OnSubscribeFunc<T> using( final Func0<RESOURCE> resourceFactory, final Func1<RESOURCE, Observable<T>> observableFactory) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { Subscription resourceSubscription = Subscriptions.empty(); try { RESOURCE resource = resourceFactory.call(); if (resource != null) { resourceSubscription = resource; } Observable<T> observable = observableFactory.call(resource); SafeObservableSubscription subscription = new SafeObservableSubscription(); return subscription.wrap(new CompositeSubscription( observable.subscribe(new SafeObserver<T>( subscription, observer)), resourceSubscription)); } catch (Throwable e) { resourceSubscription.unsubscribe(); return Observable.<T> error(e).subscribe(observer); } } }; }  }
OperationUsing { public static <T, RESOURCE extends Subscription> OnSubscribeFunc<T> using( final Func0<RESOURCE> resourceFactory, final Func1<RESOURCE, Observable<T>> observableFactory) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { Subscription resourceSubscription = Subscriptions.empty(); try { RESOURCE resource = resourceFactory.call(); if (resource != null) { resourceSubscription = resource; } Observable<T> observable = observableFactory.call(resource); SafeObservableSubscription subscription = new SafeObservableSubscription(); return subscription.wrap(new CompositeSubscription( observable.subscribe(new SafeObserver<T>( subscription, observer)), resourceSubscription)); } catch (Throwable e) { resourceSubscription.unsubscribe(); return Observable.<T> error(e).subscribe(observer); } } }; }  static OnSubscribeFunc<T> using(
            final Func0<RESOURCE> resourceFactory,
            final Func1<RESOURCE, Observable<T>> observableFactory); }
OperationUsing { public static <T, RESOURCE extends Subscription> OnSubscribeFunc<T> using( final Func0<RESOURCE> resourceFactory, final Func1<RESOURCE, Observable<T>> observableFactory) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { Subscription resourceSubscription = Subscriptions.empty(); try { RESOURCE resource = resourceFactory.call(); if (resource != null) { resourceSubscription = resource; } Observable<T> observable = observableFactory.call(resource); SafeObservableSubscription subscription = new SafeObservableSubscription(); return subscription.wrap(new CompositeSubscription( observable.subscribe(new SafeObserver<T>( subscription, observer)), resourceSubscription)); } catch (Throwable e) { resourceSubscription.unsubscribe(); return Observable.<T> error(e).subscribe(observer); } } }; }  static OnSubscribeFunc<T> using(
            final Func0<RESOURCE> resourceFactory,
            final Func1<RESOURCE, Observable<T>> observableFactory);  }