@Test public void testGetLevel1() throws Exception { String key = "TEST_KEY"; String value = "TEST_VALUE"; SimpleObject object = new SimpleObject(value); when(cache1.get(eq(key), eq(SimpleObject.class))).thenReturn(Observable.just(object)); Observable<SimpleObject> observable = waterfallCache.get(key, SimpleObject.class); ObservableTestUtils.testObservable(observable, simpleObject -> assertEquals(value, simpleObject.getValue())); verify(cache1).get(eq(key), eq(SimpleObject.class)); verifyZeroInteractions(cache2); }
@Override public <T> Observable<T> get(final String key, final Type typeOfT) { if (memoryCache != null) { T memoryValue = (T) memoryCache.get(key); if (memoryValue != null) { return Observable.just(memoryValue).compose(applySchedulers()); } } return achieveOnce(null, cache -> cache.<T>get(key, typeOfT), value -> value != null) .flatMap(resultWrapper -> { if (resultWrapper.result != null && resultWrapper.hitCacheIdx > 0) { Observable<Boolean> observable = Observable.just(false); for (int i = 0; i < resultWrapper.hitCacheIdx; i++) { RxCache cache = caches.get(i); observable = observable.flatMap(success -> cache.put(key, resultWrapper.result)); } return observable.map(success -> resultWrapper.result); } return Observable.just(resultWrapper.result); }); }
WaterfallCache implements Cache { @Override public <T> Observable<T> get(final String key, final Type typeOfT) { if (memoryCache != null) { T memoryValue = (T) memoryCache.get(key); if (memoryValue != null) { return Observable.just(memoryValue).compose(applySchedulers()); } } return achieveOnce(null, cache -> cache.<T>get(key, typeOfT), value -> value != null) .flatMap(resultWrapper -> { if (resultWrapper.result != null && resultWrapper.hitCacheIdx > 0) { Observable<Boolean> observable = Observable.just(false); for (int i = 0; i < resultWrapper.hitCacheIdx; i++) { RxCache cache = caches.get(i); observable = observable.flatMap(success -> cache.put(key, resultWrapper.result)); } return observable.map(success -> resultWrapper.result); } return Observable.just(resultWrapper.result); }); } }
WaterfallCache implements Cache { @Override public <T> Observable<T> get(final String key, final Type typeOfT) { if (memoryCache != null) { T memoryValue = (T) memoryCache.get(key); if (memoryValue != null) { return Observable.just(memoryValue).compose(applySchedulers()); } } return achieveOnce(null, cache -> cache.<T>get(key, typeOfT), value -> value != null) .flatMap(resultWrapper -> { if (resultWrapper.result != null && resultWrapper.hitCacheIdx > 0) { Observable<Boolean> observable = Observable.just(false); for (int i = 0; i < resultWrapper.hitCacheIdx; i++) { RxCache cache = caches.get(i); observable = observable.flatMap(success -> cache.put(key, resultWrapper.result)); } return observable.map(success -> resultWrapper.result); } return Observable.just(resultWrapper.result); }); } private  WaterfallCache(List<RxCache> caches, Scheduler observeOnScheduler, int inlineMemoryCacheSize); }
WaterfallCache implements Cache { @Override public <T> Observable<T> get(final String key, final Type typeOfT) { if (memoryCache != null) { T memoryValue = (T) memoryCache.get(key); if (memoryValue != null) { return Observable.just(memoryValue).compose(applySchedulers()); } } return achieveOnce(null, cache -> cache.<T>get(key, typeOfT), value -> value != null) .flatMap(resultWrapper -> { if (resultWrapper.result != null && resultWrapper.hitCacheIdx > 0) { Observable<Boolean> observable = Observable.just(false); for (int i = 0; i < resultWrapper.hitCacheIdx; i++) { RxCache cache = caches.get(i); observable = observable.flatMap(success -> cache.put(key, resultWrapper.result)); } return observable.map(success -> resultWrapper.result); } return Observable.just(resultWrapper.result); }); } private  WaterfallCache(List<RxCache> caches, Scheduler observeOnScheduler, int inlineMemoryCacheSize); @Override Observable<T> get(final String key, final Type typeOfT); @Override Observable<Boolean> put(final String key, final Object object); @Override Observable<Boolean> contains(final String key); @Override Observable<Boolean> remove(final String key); @Override Observable<Boolean> clear(); @Override void getAsync(String key, Type typeOfT, WaterfallGetCallback<T> callback); @Override void putAsync(String key, Object object, WaterfallCallback callback); @Override void containsAsync(String key, WaterfallGetCallback<Boolean> callback); @Override void removeAsync(String key, WaterfallCallback callback); @Override void clearAsync(WaterfallCallback callback); void setObserveOnScheduler(Scheduler scheduler); static Builder builder(); }
WaterfallCache implements Cache { @Override public <T> Observable<T> get(final String key, final Type typeOfT) { if (memoryCache != null) { T memoryValue = (T) memoryCache.get(key); if (memoryValue != null) { return Observable.just(memoryValue).compose(applySchedulers()); } } return achieveOnce(null, cache -> cache.<T>get(key, typeOfT), value -> value != null) .flatMap(resultWrapper -> { if (resultWrapper.result != null && resultWrapper.hitCacheIdx > 0) { Observable<Boolean> observable = Observable.just(false); for (int i = 0; i < resultWrapper.hitCacheIdx; i++) { RxCache cache = caches.get(i); observable = observable.flatMap(success -> cache.put(key, resultWrapper.result)); } return observable.map(success -> resultWrapper.result); } return Observable.just(resultWrapper.result); }); } private  WaterfallCache(List<RxCache> caches, Scheduler observeOnScheduler, int inlineMemoryCacheSize); @Override Observable<T> get(final String key, final Type typeOfT); @Override Observable<Boolean> put(final String key, final Object object); @Override Observable<Boolean> contains(final String key); @Override Observable<Boolean> remove(final String key); @Override Observable<Boolean> clear(); @Override void getAsync(String key, Type typeOfT, WaterfallGetCallback<T> callback); @Override void putAsync(String key, Object object, WaterfallCallback callback); @Override void containsAsync(String key, WaterfallGetCallback<Boolean> callback); @Override void removeAsync(String key, WaterfallCallback callback); @Override void clearAsync(WaterfallCallback callback); void setObserveOnScheduler(Scheduler scheduler); static Builder builder();  }