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);  }
@Test public void testUsing() { final Resource resource = mock(Resource.class); when(resource.getTextFromWeb()).thenReturn("Hello world!"); Func0<Resource> resourceFactory = new Func0<Resource>() { @Override public Resource call() { return resource; } }; Func1<Resource, Observable<String>> observableFactory = new Func1<Resource, Observable<String>>() { @Override public Observable<String> call(Resource resource) { return Observable.from(resource.getTextFromWeb().split(" ")); } }; @SuppressWarnings("unchecked") Observer<String> observer = (Observer<String>) mock(Observer.class); Observable<String> observable = Observable.create(using( resourceFactory, observableFactory)); observable.subscribe(observer); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onNext("Hello"); inOrder.verify(observer, times(1)).onNext("world!"); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(resource, times(1)).unsubscribe(); }
@Test public void testUsingWithSubscribingTwice() { Func0<Resource> resourceFactory = new Func0<Resource>() { @Override public Resource call() { return new Resource() { boolean first = true; @Override public String getTextFromWeb() { if (first) { first = false; return "Hello world!"; } return "Nothing"; } @Override public void unsubscribe() { } @Override public boolean isUnsubscribed() { return false; } }; } }; Func1<Resource, Observable<String>> observableFactory = new Func1<Resource, Observable<String>>() { @Override public Observable<String> call(Resource resource) { return Observable.from(resource.getTextFromWeb().split(" ")); } }; @SuppressWarnings("unchecked") Observer<String> observer = (Observer<String>) mock(Observer.class); Observable<String> observable = Observable.create(using( resourceFactory, observableFactory)); observable.subscribe(observer); observable.subscribe(observer); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onNext("Hello"); inOrder.verify(observer, times(1)).onNext("world!"); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verify(observer, times(1)).onNext("Hello"); inOrder.verify(observer, times(1)).onNext("world!"); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
@Test(expected = TestException.class) public void testUsingWithResourceFactoryError() { Func0<Subscription> resourceFactory = new Func0<Subscription>() { @Override public Subscription call() { throw new TestException(); } }; Func1<Subscription, Observable<Integer>> observableFactory = new Func1<Subscription, Observable<Integer>>() { @Override public Observable<Integer> call(Subscription subscription) { return Observable.empty(); } }; Observable.create(using(resourceFactory, observableFactory)) .toBlockingObservable().last(); }
@Test public void testUsingWithObservableFactoryError() { 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) { 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(); } }
@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(); } }