@Test public void testAsync() { Observable<Integer> source = Observable.range(1, 10000); for (int i = 0; i < 100; i++) { TestObserver<Integer> to1 = new TestObserver<Integer>(); ObservableCache<Integer> cached = (ObservableCache<Integer>)ObservableCache.from(source); cached.observeOn(Schedulers.computation()).subscribe(to1); to1.awaitTerminalEvent(2, TimeUnit.SECONDS); to1.assertNoErrors(); to1.assertComplete(); assertEquals(10000, to1.values().size()); TestObserver<Integer> to2 = new TestObserver<Integer>(); cached.observeOn(Schedulers.computation()).subscribe(to2); to2.awaitTerminalEvent(2, TimeUnit.SECONDS); to2.assertNoErrors(); to2.assertComplete(); assertEquals(10000, to2.values().size()); } }
public static <T> Observable<T> from(Observable<T> source) { return from(source, 16); }
ObservableCache extends AbstractObservableWithUpstream<T, T> { public static <T> Observable<T> from(Observable<T> source) { return from(source, 16); } }
ObservableCache extends AbstractObservableWithUpstream<T, T> { public static <T> Observable<T> from(Observable<T> source) { return from(source, 16); } private  ObservableCache(Observable<T> source, CacheState<T> state); }
ObservableCache extends AbstractObservableWithUpstream<T, T> { public static <T> Observable<T> from(Observable<T> source) { return from(source, 16); } private  ObservableCache(Observable<T> source, CacheState<T> state); static Observable<T> from(Observable<T> source); static Observable<T> from(Observable<T> source, int capacityHint); }
ObservableCache extends AbstractObservableWithUpstream<T, T> { public static <T> Observable<T> from(Observable<T> source) { return from(source, 16); } private  ObservableCache(Observable<T> source, CacheState<T> state); static Observable<T> from(Observable<T> source); static Observable<T> from(Observable<T> source, int capacityHint);  }