@Test public void testCache() throws InterruptedException { final AtomicInteger counter = new AtomicInteger(); Observable<String> o = Observable.create(cache(Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(final Observer<? super String> observer) { final BooleanSubscription subscription = new BooleanSubscription(); new Thread(new Runnable() { @Override public void run() { counter.incrementAndGet(); System.out.println("published observable being executed"); observer.onNext("one"); observer.onCompleted(); } }).start(); return subscription; } }))); final CountDownLatch latch = new CountDownLatch(2); o.subscribe(new Action1<String>() { @Override public void call(String v) { assertEquals("one", v); System.out.println("v: " + v); latch.countDown(); } }); o.subscribe(new Action1<String>() { @Override public void call(String v) { assertEquals("one", v); System.out.println("v: " + v); latch.countDown(); } }); if (!latch.await(1000, TimeUnit.MILLISECONDS)) { fail("subscriptions did not receive values"); } assertEquals(1, counter.get()); }
public static <T> OnSubscribeFunc<T> cache(final Observable<? extends T> source) { return new OnSubscribeFunc<T>() { final AtomicBoolean subscribed = new AtomicBoolean(false); private final ReplaySubject<T> cache = ReplaySubject.create(); @Override public Subscription onSubscribe(Observer<? super T> observer) { if (subscribed.compareAndSet(false, true)) { source.subscribe(cache); } return cache.subscribe(observer); } }; }
OperationCache { public static <T> OnSubscribeFunc<T> cache(final Observable<? extends T> source) { return new OnSubscribeFunc<T>() { final AtomicBoolean subscribed = new AtomicBoolean(false); private final ReplaySubject<T> cache = ReplaySubject.create(); @Override public Subscription onSubscribe(Observer<? super T> observer) { if (subscribed.compareAndSet(false, true)) { source.subscribe(cache); } return cache.subscribe(observer); } }; } }
OperationCache { public static <T> OnSubscribeFunc<T> cache(final Observable<? extends T> source) { return new OnSubscribeFunc<T>() { final AtomicBoolean subscribed = new AtomicBoolean(false); private final ReplaySubject<T> cache = ReplaySubject.create(); @Override public Subscription onSubscribe(Observer<? super T> observer) { if (subscribed.compareAndSet(false, true)) { source.subscribe(cache); } return cache.subscribe(observer); } }; }  }
OperationCache { public static <T> OnSubscribeFunc<T> cache(final Observable<? extends T> source) { return new OnSubscribeFunc<T>() { final AtomicBoolean subscribed = new AtomicBoolean(false); private final ReplaySubject<T> cache = ReplaySubject.create(); @Override public Subscription onSubscribe(Observer<? super T> observer) { if (subscribed.compareAndSet(false, true)) { source.subscribe(cache); } return cache.subscribe(observer); } }; }  static OnSubscribeFunc<T> cache(final Observable<? extends T> source); }
OperationCache { public static <T> OnSubscribeFunc<T> cache(final Observable<? extends T> source) { return new OnSubscribeFunc<T>() { final AtomicBoolean subscribed = new AtomicBoolean(false); private final ReplaySubject<T> cache = ReplaySubject.create(); @Override public Subscription onSubscribe(Observer<? super T> observer) { if (subscribed.compareAndSet(false, true)) { source.subscribe(cache); } return cache.subscribe(observer); } }; }  static OnSubscribeFunc<T> cache(final Observable<? extends T> source);  }