Обработка null в RxJava2

С предстоящей RxJava2 release одним из важных изменений является то, что null больше не принимается как элемент потока, то есть следующий код будет исключение: Observable.just(null)

Честно говоря, у меня смешанные чувства по поводу этого изменения, и часть меня понимает, что он будет применять чистые API-интерфейсы, но я могу видеть ряд вариантов использования, когда это может быть проблемой.

Например, в моем приложении у меня есть кеш в памяти:

@Nullable CacheItem findCacheItem(long id);

CacheItem может отсутствовать в кеше, поэтому метод может возвращать нулевое значение.

Способ, которым он используется с Rx * -, выглядит следующим образом:

Observable<CacheItem> getStream(final long id) {
    return Observable.fromCallable(new Callable<CacheItem>() {
        @Override public CacheItem call() throws Exception {
            return findCacheItem(id);
        }
    });
}

Таким образом, при таком подходе я могу получить null в моем потоке, который является полностью допустимой ситуацией, поэтому он правильно обрабатывается на стороне получателя - пусть пользовательский интерфейс изменяет свое состояние, если элемент отсутствует в кеше:

Observable.just(user)
          .map(user -> user.getName())
          .map(name -> convertNameToId(name))
          .flatMap(id -> getStream(id))
          .map(cacheItem -> getUserInfoFromCacheItem(cacheItem))
          .subscribe(
              userInfo -> {
                  if(userInfo != null) showUserInfo();
                  else showPrompt();
              }
          );

С RxJava2 мне больше не разрешается отправлять null вниз по потоку, поэтому мне нужно либо обернуть мой CacheItem в какой-то другой класс, либо сделать мой поток, создав эту оболочку, или сделать довольно большие архитектурные изменения.

Обтекание каждого элемента потока в NULL-эквиваленте не выглядит правильным.

Я пропустил что-то фундаментальное здесь?

Похоже, что такая ситуация, как моя, довольно популярна, поэтому мне интересно, какая рекомендуемая стратегия для решения этой проблемы задана новой политикой "нет нуля" в RxJava2?

ИЗМЕНИТЬ См. Последующий разговор в RxJava GitHub repo

Ответ 1

Ну, есть несколько способов представить то, что вы хотите.

Один из вариантов - использовать Observable<Optional<CacheItem>>:

Observable<Optional<CacheItem>> getStream(final long id) {
  return Observable.defer(() -> {
    return Observable.just(Optional.ofNullable(findCacheItem(id)));
  });
}

public static <T> Transformer<Optional<T>, T> deoptionalize() {
  return src -> 
      src.flatMap(item -> item.isPresent()
             ? Observable.just(item.get())
             : Observable.empty();
}

Затем вы используете .compose(deoptionalize()) для сопоставления от необязательного к необязательному Observable.

Ответ 2

В качестве другого решения вы можете добавить статический экземпляр CacheItem.NULL и вернуть его подписчику, когда нет кэшированных данных

Single
    .concat(loadFromMemory(), loadFromDb(), loadFromServer())
    .takeFirst { it != CachedItem.NULL }
    .subscribe(

Ответ 3

Вы можете использовать RxJava2-Nullable для обработки нулевого значения в RxJava2.

В вашей ситуации вы можете:

Observable<CacheItem> getStream(final long id) {
    return RxNullable.fromCallable(() -> findCacheItem(id))
                     .onNullDrop()
                     .observable();
}

Чтобы вызвать showPrompt, когда он равен null, вы можете сделать:

Observable.just(user)
          .map(user -> user.getName())
          .map(name -> convertNameToId(name))
          .flatMap(id -> getStream(id).onNullRun(() -> showPrompt()))
          .map(cacheItem -> getUserInfoFromCacheItem(cacheItem))
          .subscribe(userInfo -> showUserInfo());

NullableObservable<CacheItem> getStream(final long id) {
    return RxNullable.fromCallable(() -> findCacheItem(id)).observable();
}

Ответ 4

Возможное решение - использовать Maybe.switchIfEmpty

Пример:

public static <T> Maybe<T> maybeOfNullable(T value) {
    return value == null ? Maybe.empty() : Maybe.just(value);
}

maybeOfNullable(user)
        .map(user -> user.getName())
        .map(name -> convertNameToId(name))
        .flatMap(id -> getStream(id))
        .map(cacheItem -> getUserInfoFromCacheItem(cacheItem))
        // perform another action in case there are no any non null item emitted
        .switchIfEmpty(Maybe.fromAction(() -> showPrompt()))
        .subscribe(userInfo -> showUserInfo());