Wrapped nullable String returned to onNext with an Optional to avoid NPE...

Wrapped nullable String returned to onNext with an Optional to avoid NPE eventually happening since RxJava 2.x does not support it

app/src/main/java/chat/rocket/android/RocketChatCache.java

@@ -3,11 +3,13 @@ package chat.rocket.android;
 import android.content.Context;
 import android.content.SharedPreferences;
 
-import io.reactivex.BackpressureStrategy;
-import io.reactivex.Flowable;
+import com.hadisatrio.optional.Optional;
 
 import java.util.UUID;
 
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+
 /**
  * sharedpreference-based cache.
  */
@@ -51,11 +53,11 @@ public class RocketChatCache {
     return preferences.getString(KEY_PUSH_ID, null);
   }
 
-  public Flowable<String> getSelectedServerHostnamePublisher() {
+  public Flowable<Optional<String>> getSelectedServerHostnamePublisher() {
     return getValuePublisher(KEY_SELECTED_SERVER_HOSTNAME);
   }
 
-  public Flowable<String> getSelectedRoomIdPublisher() {
+  public Flowable<Optional<String>> getSelectedRoomIdPublisher() {
     return getValuePublisher(KEY_SELECTED_ROOM_ID);
   }
 
@@ -75,12 +77,13 @@ public class RocketChatCache {
     getEditor().putString(key, value).apply();
   }
 
-  private Flowable<String> getValuePublisher(final String key) {
+  private Flowable<Optional<String>> getValuePublisher(final String key) {
     return Flowable.create(emitter -> {
       SharedPreferences.OnSharedPreferenceChangeListener
           listener = (sharedPreferences, changedKey) -> {
         if (key.equals(changedKey) && !emitter.isCancelled()) {
-          emitter.onNext(getString(key, null));
+          String value = getString(key, null);
+          emitter.onNext(Optional.of(value));
         }
       };
 

app/src/main/java/chat/rocket/android/activity/AbstractAuthedActivity.java

@@ -4,6 +4,8 @@ import android.content.Intent;
 import android.os.Bundle;
 import android.support.annotation.Nullable;
 
+import com.hadisatrio.optional.Optional;
+
 import java.util.List;
 
 import chat.rocket.android.LaunchUtil;
@@ -179,6 +181,7 @@ abstract class AbstractAuthedActivity extends AbstractFragmentActivity {
   private void subscribeToConfigChanges() {
     compositeDisposable.add(
         rocketChatCache.getSelectedServerHostnamePublisher()
+            .map(Optional::get)
             .distinctUntilChanged()
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
@@ -190,6 +193,7 @@ abstract class AbstractAuthedActivity extends AbstractFragmentActivity {
 
     compositeDisposable.add(
         rocketChatCache.getSelectedRoomIdPublisher()
+            .map(Optional::get)
             .distinctUntilChanged()
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())

app/src/main/java/chat/rocket/android/service/internal/AbstractRocketChatCacheObserver.java

@@ -2,6 +2,8 @@ package chat.rocket.android.service.internal;
 
 import android.content.Context;
 
+import com.hadisatrio.optional.Optional;
+
 import chat.rocket.android.log.RCLog;
 import io.reactivex.disposables.CompositeDisposable;
 
@@ -48,6 +50,7 @@ public abstract class AbstractRocketChatCacheObserver implements Registrable {
     compositeDisposable.add(
         new RocketChatCache(context)
             .getSelectedRoomIdPublisher()
+            .map(Optional::get)
             .subscribe(this::updateRoomIdWith, RCLog::e)
     );
   }