@Test public void testUnsubscribeSource() { Action0 unsubscribe = mock(Action0.class); Observable<Integer> o = Observable.just(1).doOnUnsubscribe(unsubscribe).cache(); o.subscribe(); o.subscribe(); o.subscribe(); verify(unsubscribe, times(1)).call(); }
@Override public void call(Subscriber<? super T> s) { if (SRC_SUBSCRIBED_UPDATER.compareAndSet(this, 0, 1)) { source.subscribe(cache); } cache.unsafeSubscribe(s); }
OnSubscribeCache implements OnSubscribe<T> { @Override public void call(Subscriber<? super T> s) { if (SRC_SUBSCRIBED_UPDATER.compareAndSet(this, 0, 1)) { source.subscribe(cache); } cache.unsafeSubscribe(s); } }
OnSubscribeCache implements OnSubscribe<T> { @Override public void call(Subscriber<? super T> s) { if (SRC_SUBSCRIBED_UPDATER.compareAndSet(this, 0, 1)) { source.subscribe(cache); } cache.unsafeSubscribe(s); } OnSubscribeCache(Observable<? extends T> source);  OnSubscribeCache(Observable<? extends T> source, int capacity);   OnSubscribeCache(Observable<? extends T> source, Subject<? super T, ? extends T> cache); }
OnSubscribeCache implements OnSubscribe<T> { @Override public void call(Subscriber<? super T> s) { if (SRC_SUBSCRIBED_UPDATER.compareAndSet(this, 0, 1)) { source.subscribe(cache); } cache.unsafeSubscribe(s); } OnSubscribeCache(Observable<? extends T> source);  OnSubscribeCache(Observable<? extends T> source, int capacity);   OnSubscribeCache(Observable<? extends T> source, Subject<? super T, ? extends T> cache); @Override void call(Subscriber<? super T> s); }
OnSubscribeCache implements OnSubscribe<T> { @Override public void call(Subscriber<? super T> s) { if (SRC_SUBSCRIBED_UPDATER.compareAndSet(this, 0, 1)) { source.subscribe(cache); } cache.unsafeSubscribe(s); } OnSubscribeCache(Observable<? extends T> source);  OnSubscribeCache(Observable<? extends T> source, int capacity);   OnSubscribeCache(Observable<? extends T> source, Subject<? super T, ? extends T> cache); @Override void call(Subscriber<? super T> s);  }