在 RxJava2 中处理 null

Question

随着即将到来的RxJava2 release 的重要变化之一是null 不再被接受为流元素，即以下代码将引发异常：Observable.just(null)

老实说，我对这一变化有复杂的感受，我的一部分理解它将强制执行干净的 API，但我可以看到许多用例可能会成为问题。

例如，在我的应用中，我有一个内存缓存：

@Nullable CacheItem findCacheItem(long id);

CacheItem 可能不存在于缓存中，因此方法可能返回 null 值。

与 Rx* 一起使用的方式如下：

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

因此，使用这种方法，我的流中可能会得到 null，这是完全有效的情况，因此在接收端可以正确处理 - 假设如果缓存中不存在项目，UI 会更改其状态：

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 包装到其他类中并让我的流生成该包装器，或者进行相当大的架构更改。

将每个单独的流元素包装成可以为空的对应元素对我来说并不合适。

我在这里遗漏了一些基本的东西吗？

似乎像我这样的情况很流行，所以我很好奇鉴于 RxJava2 中新的“no null”策略来解决这个问题的推荐策略是什么？

编辑 请看RxJava GitHub repo的后续对话

Answer 1

Nulls 是奇怪的东西。从一方面来看，它们并不像具体价值那样标志着完全成功。另一方面，它们并不像Exception 那样表示完全失败。所以可以说null 介于快乐路径和失败路径之间。每当null 被禁止时，它都会迫使您“下定决心”并将其移至其中一个组。要么使它完全成功（通过将其包装到某种对象中并将空处理逻辑放入此对象），要么完全错误并将其作为Exception 抛出。

对我来说，限制通往完全成功的快乐之路并处理onError 频道中的所有偏差非常有益：

Observable.just(user)
      .map(user -> user.getName())
      .map(name -> convertNameToId(name))
      .flatMap(id -> getStream(id))
      .map(cacheItem -> getUserInfoFromCacheItem(cacheItem)) //throws custom exception
      .subscribe(
          userInfo -> {
              showUserInfo(userInfo);
          },
          ex -> {
              showPrompt(ex); //handles the custom exception
          }
      );

最好的部分是onError 将所有转换（通常取决于值是否为非空）一直短路到您决定可以处理它的地步。要么在最后的subscribe/onError，要么在链中间的某个地方onErrorResumeNext。请记住，允许onError 一直传播会终止链。

Answer 2

可能的解决方案是使用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());

Answer 3

您可以使用RxJava2-Nullable 在 RxJava2 中处理空值。

针对你的情况，你可以这样做：

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

要在 showPrompt 为空时调用它，您可以这样做：

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();
}

Answer 4

嗯，有几种方法可以表示你想要的。

一种选择是使用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 映射到非可选的 Observable。

Answer 5

作为另一种解决方案，您可以添加静态实例CacheItem.NULL，并在没有缓存数据时将其返回给订阅者

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