@Test public void put_WhenMergeThrows_Errors() { memoryStoreCore = new MemoryStoreCore<>((s, s2) -> { throw new IllegalStateException("Mock"); }); memoryStoreCore.put(100, "test value 1"); memoryStoreCore.put(100, "test value 2") .test() .assertError(IllegalStateException.class); }
@NonNull @Override public Single<Boolean> put(@NonNull final T id, @NonNull final U item) { checkNotNull(id); checkNotNull(item); final int hash = getHashCodeForId(id); U newItem = item; boolean valuesEqual = false; if (cache.containsKey(hash)) { final U currentItem = cache.get(hash); valuesEqual = newItem.equals(currentItem); if (!valuesEqual) { Log.v(TAG, "Merging values at " + id); try { newItem = putMergeFunction.apply(currentItem, newItem); } catch (Exception e) { return Single.error(e); } valuesEqual = newItem.equals(currentItem); } } if (valuesEqual) { Log.v(TAG, "Data already up to date at " + id); return Single.just(false); } cache.put(hash, newItem); subject.onNext(new StoreItem<>(id, newItem)); if (subjectCache.containsKey(hash)) { subjectCache.get(hash).onNext(newItem); } return Single.just(true); }
MemoryStoreCore implements StoreCoreInterface<T, U> { @NonNull @Override public Single<Boolean> put(@NonNull final T id, @NonNull final U item) { checkNotNull(id); checkNotNull(item); final int hash = getHashCodeForId(id); U newItem = item; boolean valuesEqual = false; if (cache.containsKey(hash)) { final U currentItem = cache.get(hash); valuesEqual = newItem.equals(currentItem); if (!valuesEqual) { Log.v(TAG, "Merging values at " + id); try { newItem = putMergeFunction.apply(currentItem, newItem); } catch (Exception e) { return Single.error(e); } valuesEqual = newItem.equals(currentItem); } } if (valuesEqual) { Log.v(TAG, "Data already up to date at " + id); return Single.just(false); } cache.put(hash, newItem); subject.onNext(new StoreItem<>(id, newItem)); if (subjectCache.containsKey(hash)) { subjectCache.get(hash).onNext(newItem); } return Single.just(true); } }
MemoryStoreCore implements StoreCoreInterface<T, U> { @NonNull @Override public Single<Boolean> put(@NonNull final T id, @NonNull final U item) { checkNotNull(id); checkNotNull(item); final int hash = getHashCodeForId(id); U newItem = item; boolean valuesEqual = false; if (cache.containsKey(hash)) { final U currentItem = cache.get(hash); valuesEqual = newItem.equals(currentItem); if (!valuesEqual) { Log.v(TAG, "Merging values at " + id); try { newItem = putMergeFunction.apply(currentItem, newItem); } catch (Exception e) { return Single.error(e); } valuesEqual = newItem.equals(currentItem); } } if (valuesEqual) { Log.v(TAG, "Data already up to date at " + id); return Single.just(false); } cache.put(hash, newItem); subject.onNext(new StoreItem<>(id, newItem)); if (subjectCache.containsKey(hash)) { subjectCache.get(hash).onNext(newItem); } return Single.just(true); } MemoryStoreCore();  MemoryStoreCore(@NonNull final BiFunction<U, U, U> putMergeFunction); }
MemoryStoreCore implements StoreCoreInterface<T, U> { @NonNull @Override public Single<Boolean> put(@NonNull final T id, @NonNull final U item) { checkNotNull(id); checkNotNull(item); final int hash = getHashCodeForId(id); U newItem = item; boolean valuesEqual = false; if (cache.containsKey(hash)) { final U currentItem = cache.get(hash); valuesEqual = newItem.equals(currentItem); if (!valuesEqual) { Log.v(TAG, "Merging values at " + id); try { newItem = putMergeFunction.apply(currentItem, newItem); } catch (Exception e) { return Single.error(e); } valuesEqual = newItem.equals(currentItem); } } if (valuesEqual) { Log.v(TAG, "Data already up to date at " + id); return Single.just(false); } cache.put(hash, newItem); subject.onNext(new StoreItem<>(id, newItem)); if (subjectCache.containsKey(hash)) { subjectCache.get(hash).onNext(newItem); } return Single.just(true); } MemoryStoreCore();  MemoryStoreCore(@NonNull final BiFunction<U, U, U> putMergeFunction); @NonNull @Override Maybe<U> getCached(@NonNull final T id); @NonNull @Override Single<List<U>> getCached(); @NonNull @Override Observable<U> getStream(@NonNull final T id); @NonNull @Override Observable<U> getStream(); @NonNull @Override Single<Boolean> put(@NonNull final T id, @NonNull final U item); @NonNull @Override Single<Boolean> delete(@NonNull final T id); }
MemoryStoreCore implements StoreCoreInterface<T, U> { @NonNull @Override public Single<Boolean> put(@NonNull final T id, @NonNull final U item) { checkNotNull(id); checkNotNull(item); final int hash = getHashCodeForId(id); U newItem = item; boolean valuesEqual = false; if (cache.containsKey(hash)) { final U currentItem = cache.get(hash); valuesEqual = newItem.equals(currentItem); if (!valuesEqual) { Log.v(TAG, "Merging values at " + id); try { newItem = putMergeFunction.apply(currentItem, newItem); } catch (Exception e) { return Single.error(e); } valuesEqual = newItem.equals(currentItem); } } if (valuesEqual) { Log.v(TAG, "Data already up to date at " + id); return Single.just(false); } cache.put(hash, newItem); subject.onNext(new StoreItem<>(id, newItem)); if (subjectCache.containsKey(hash)) { subjectCache.get(hash).onNext(newItem); } return Single.just(true); } MemoryStoreCore();  MemoryStoreCore(@NonNull final BiFunction<U, U, U> putMergeFunction); @NonNull @Override Maybe<U> getCached(@NonNull final T id); @NonNull @Override Single<List<U>> getCached(); @NonNull @Override Observable<U> getStream(@NonNull final T id); @NonNull @Override Observable<U> getStream(); @NonNull @Override Single<Boolean> put(@NonNull final T id, @NonNull final U item); @NonNull @Override Single<Boolean> delete(@NonNull final T id);  }