在RxJava2中处理null

Question

随着即将推出的RxJava2版本的一个重要变化是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中给出新的"无空"政策时,解决这个问题的建议策略是什么？

编辑 请参阅RxJava GitHub仓库中的后续对话

Answer 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.