@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(); }
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);  }