Как я могу кешировать элементы в RxJava и избежать паники?

Question

Допустим, у меня есть следующий код:

Entity getEntity(GUID entityId) {
  Entity entity = entityLRUCache.get(entityId);
  if (entity == null) {
    entity = longLoadFromDatabase(entityId);
    entityLRUCache.put(entityId, entity);
  }
  return entity;
}

Благодаря @BenManes я могу использовать Caffeine для решения проблем с кэшем:

Entity getEntity(GUID entityId) {
  return entityCache.get(entityId, this::longLoadFromDatabase);
}

Но сейчас метод longLoadFromDatabase возвращает Single<Entity> вместо самой сущности (этот второй аргумент для get это картограф из int -> Entity), поэтому предыдущие решения больше не будут работать.

Answer 1

Вы можете использовать SingleSubject в качестве заполнителя на параллельной карте:

ConcurrentMap<GUID, SingleSubject<Entity>> map = ...

public Single<Entity> getEntity(GUID guid) {
    SingleSubject<Entity> e = map.get(guid);
    if (e == null) {
        e = SingleSubject.create();
        SingleSubject<Entity> f = map.putIfAbsent(guid, e);
        if (f == null) {
            longLoadFromDatabase(guid).subscribe(e);
        } else {
            e = f;
        }
    }
    return e;
}

Answer 2

Комбинируя предложение @akarnokd с асинхронной поддержкой Caffeine, вы можете использовать AsyncLoadingCache с конвертерами Rx.

AsyncLoadingCache<Integer, Entity> cache = Caffeine.newBuilder()
    .buildAsync((key, executor) -> SingleInterop.get()
        .apply(longLoadFromDatabase(key)).toCompletableFuture());
...
return SingleInterop.fromFuture(cache.get(123));

Кэш автоматически удалит запись, если в будущем произойдет сбой с исключением или он будет преобразован в нулевое значение.