Цепочка RxJava, наблюдаемая для обновления / создания записи в базе данных Room

Question

Я хочу найти и обновить существующую запись в комнате с помощью RxJava. Если записи нет, следует создать новую.

Например, допустим, у меня есть следующие запросы:

@Insert
Single<Long> createContent(Content content);

@Query("SELECT * FROM Content WHERE contentId = :contentId")
Single<Content> searchContent(String contentId);

Мои цели:

1. Проверить наличие предыдущих данных и вернуть их значение
2. Если записи нет, создайте новую и верните ее значение

Проблема с этим подходом:

1. Всякий раз, когда нет записи из @Query, Single<Content> напрямую переходит к error, игнорируя любую карту / flatMap оператор
2. Запрос @Insert возвращает Single<Long>, а @Query возвращает Single<Content>

Можно ли как-нибудь вызвать и вернуть новый Observable из-за ошибки? Примерно так:

daoAccess.searchContent(contentId)
                .subscribeOn(Schedulers.io())
                .map(Resource::success)
                .onErrorResumeNext(new Function<Throwable, Single<Content>>() {
                    @Override
                    public Single<Content> apply(Throwable throwable) throws Exception {
                        return daoAccess.createContent(contentId);
                    }
                })

Answer 1

Вы можете использовать Single.onErrorResumeNext():

daoAccess.searchContent(contentId)
         .subscribeOn(Schedulers.io())
         .map(Resource::success)
         .onErrorResumeNext(throwable ->
             daoAccess.createContent(content)
                      .map(id -> Resource.success(content))
         )