Handling null in RxJava2

Question

With the upcoming RxJava2 release one of the important changes is that null is no longer accepted as a stream element, i.e. following code will throw an exception: Observable.just(null)

Honestly, I have mixed feelings about this change and part of me understands that it will enforce clean APIs, but I can see a number of use cases when this might be a problem.

For instance, in my app I have an in-memory cache:

@Nullable CacheItem findCacheItem(long id); 

CacheItem might not be present in cache, so method might return null value.

The way it is used with Rx* - is as following:

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

So with this approach, I might get null in my stream which is totally valid situation, so it is handled properly on receiving side - let's say UI changes its state if item is not present in cache:

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

With RxJava2 I am no longer allowed to post null down the stream, so I either need to wrap my CacheItem into some other class and make my stream produce that wrapper instead or make quite big architectural changes.

Wrapping every single stream element into nullable counterpart doesn't look right to me.

Am I missing something fundamental here?

It seems like the situation like mine is quite popular, so Im curious what is the recommended strategy to tackle this problem given new "no null" policy in RxJava2?

EDIT Please see follow-up conversation in RxJava GitHub repo

Answer 1

Well, there are several ways to represent what you want.

One option is to use 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(); } 

You then use .compose(deoptionalize()) to map from the optional to the non-optional Observable.

Answer 2

As another solution you can add static instance CacheItem.NULL, and return it to subscriber when there is no cached data

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