Some refactoring

app/src/main/java/chat/rocket/android/service/RealmBasedConnectivityManager.java

 package chat.rocket.android.service;
 
+import android.annotation.SuppressLint;
 import android.content.ComponentName;
 import android.content.Context;
 import android.content.ServiceConnection;
@@ -68,6 +69,7 @@ import io.reactivex.subjects.PublishSubject;
     }
   }
 
+  @SuppressLint("RxLeakedSubscription")
   @DebugLog
   @Override
   public void ensureConnections() {
@@ -84,6 +86,7 @@ import io.reactivex.subjects.PublishSubject;
           });
   }
 
+  @SuppressLint("RxLeakedSubscription")
   @Override
   public void addOrUpdateServer(String hostname, @Nullable String name, boolean insecure) {
     RealmBasedServerInfo.addOrUpdate(hostname, name, insecure);
@@ -95,6 +98,7 @@ import io.reactivex.subjects.PublishSubject;
         }, RCLog::e);
   }
 
+  @SuppressLint("RxLeakedSubscription")
   @Override
   public void removeServer(String hostname) {
     RealmBasedServerInfo.remove(hostname);

app/src/main/java/chat/rocket/android/service/RocketChatWebSocketThread.java

@@ -72,8 +72,8 @@ public class RocketChatWebSocketThread extends HandlerThread {
     private final RealmHelper realmHelper;
     private final ConnectivityManagerInternal connectivityManager;
     private final ArrayList<Registrable> listeners = new ArrayList<>();
-  private final CompositeDisposable hearbeatDisposable = new CompositeDisposable();
-  private final CompositeDisposable reconnectSubscription = new CompositeDisposable();
+    private final CompositeDisposable heartbeatDisposable = new CompositeDisposable();
+    private final CompositeDisposable reconnectDisposable = new CompositeDisposable();
     private boolean listenersRegistered;
 
     private static class KeepAliveTimer {
@@ -109,14 +109,13 @@ public class RocketChatWebSocketThread extends HandlerThread {
      */
     @DebugLog
     public static Single<RocketChatWebSocketThread> getStarted(Context appContext, String hostname) {
-    return Single.<RocketChatWebSocketThread>fromPublisher(objectSingleEmitter -> {
+        return Single.<RocketChatWebSocketThread>create(objectSingleEmitter -> {
             new RocketChatWebSocketThread(appContext, hostname) {
                 @Override
                 protected void onLooperPrepared() {
                     try {
                         super.onLooperPrepared();
-            objectSingleEmitter.onNext(this);
-            objectSingleEmitter.onComplete();
+                        objectSingleEmitter.onSuccess(this);
                     } catch (Exception exception) {
                         objectSingleEmitter.onError(exception);
                     }
@@ -129,6 +128,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
     @Override
     protected void onLooperPrepared() {
         super.onLooperPrepared();
+        forceInvalidateTokens();
     }
 
     private void forceInvalidateTokens() {
@@ -150,15 +150,14 @@ public class RocketChatWebSocketThread extends HandlerThread {
     @DebugLog
     public Single<Boolean> terminate() {
         if (isAlive()) {
-      return Single.fromPublisher(emitter -> {
+            return Single.create(emitter -> {
                 new Handler(getLooper()).post(() -> {
                     RCLog.d("thread %s: terminated()", Thread.currentThread().getId());
                     unregisterListenersAndClose();
                     connectivityManager.notifyConnectionLost(hostname,
                             DDPClient.REASON_CLOSED_BY_USER);
                     RocketChatWebSocketThread.super.quit();
-          emitter.onNext(true);
-          emitter.onComplete();
+                    emitter.onSuccess(true);
                 });
             });
         } else {
@@ -198,7 +197,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
         }
         keepAliveTimer.update();
 
-    return Single.fromPublisher(emitter -> {
+        return Single.create(emitter -> {
             new Thread() {
                 @Override
                 public void run() {
@@ -211,8 +210,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
                             emitter.onError(error);
                         } else {
                             keepAliveTimer.update();
-              emitter.onNext(true);
-              emitter.onComplete();
+                            emitter.onSuccess(true);
                         }
                         return null;
                     });
@@ -239,24 +237,11 @@ public class RocketChatWebSocketThread extends HandlerThread {
                 });
     }
 
-  private Single<Boolean> prepareDDPClient() {
-    // TODO: temporarily replaced checkIfConnectionAlive() call for this single checking if ddpClient is
-    // null or not. In case it is, build a new client, otherwise just keep connecting with existing one.
-    return Single.just(DDPClient.get() != null)
-        .doOnSuccess(alive -> {
-          if (!alive) {
-            RCLog.d("DDPClient#build");
-          }
-        });
-  }
-
     private Single<Boolean> connectDDPClient() {
-    return prepareDDPClient()
-        .flatMap(_val -> Single.fromPublisher(emitter -> {
+        return Single.create(emitter -> {
             ServerInfo info = connectivityManager.getServerInfoForHost(hostname);
             if (info == null) {
-            emitter.onNext(false);
-            emitter.onComplete();
+                emitter.onSuccess(false);
                 return;
             }
             RCLog.d("DDPClient#connect");
@@ -294,31 +279,29 @@ public class RocketChatWebSocketThread extends HandlerThread {
                         if (task.isFaulted()) {
                             emitter.onError(task.getError());
                         } else {
-                      emitter.onNext(true);
-                      emitter.onComplete();
+                            emitter.onSuccess(true);
                         }
                         return null;
                     });
-        }));
+        });
     }
 
     private void reconnect() {
         // if we are already trying to reconnect then return.
-    if (reconnectSubscription.size() > 0) {
+        if (reconnectDisposable.size() > 0) {
             return;
         }
+        forceInvalidateTokens();
         connectivityManager.notifyConnecting(hostname);
-    reconnectSubscription.add(
+        reconnectDisposable.add(
                 connectWithExponentialBackoff()
-            .subscribe(
-                  connected -> {
+                        .subscribe(connected -> {
                                     if (!connected) {
                                         connectivityManager.notifyConnecting(hostname);
                                     }
-                    reconnectSubscription.clear();
-                  },
-                  error -> {
-                    logErrorAndUnsubscribe(reconnectSubscription, error);
+                                    reconnectDisposable.clear();
+                                }, error -> {
+                                    logErrorAndUnsubscribe(reconnectDisposable, error);
                                     connectivityManager.notifyConnectionLost(hostname,
                                             DDPClient.REASON_NETWORK_ERROR);
                                 }
@@ -338,12 +321,11 @@ public class RocketChatWebSocketThread extends HandlerThread {
     @DebugLog
     private Single<Boolean> connect() {
         return connectDDPClient()
-        .flatMap(_val -> Single.fromPublisher(emitter -> {
+                .flatMap(_val -> Single.create(emitter -> {
                     fetchPublicSettings();
                     fetchPermissions();
                     registerListeners();
-          emitter.onNext(true);
-          emitter.onComplete();
+                    emitter.onSuccess(true);
                 }));
     }
 
@@ -387,7 +369,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
                         } else {
                             return Completable.complete();
                         }
-              }).retryWhen(RxHelper.exponentialBackoff(Integer.MAX_VALUE, 500, TimeUnit.MILLISECONDS))
+                    }).retryWhen(RxHelper.exponentialBackoff(3, 500, TimeUnit.MILLISECONDS))
                             .subscribe(
                                     () -> {
                                         createObserversAndRegister();
@@ -423,8 +405,8 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
 
     private void startHeartBeat() {
-    hearbeatDisposable.clear();
-    hearbeatDisposable.add(
+        heartbeatDisposable.clear();
+        heartbeatDisposable.add(
                 heartbeat(HEARTBEAT_PERIOD_MS)
                         .subscribe(
                                 ponged -> {
@@ -435,7 +417,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
                                 error -> {
                                     RCLog.e(error);
                                     // Stop pinging
-                      hearbeatDisposable.clear();
+                                    heartbeatDisposable.clear();
                                     if (error instanceof DDPClientCallback.Closed || error instanceof TimeoutException) {
                                         RCLog.d("Hearbeat failure: retrying connection...");
                                         reconnect();
@@ -459,7 +441,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
             registrable.unregister();
             iterator.remove();
         }
-    hearbeatDisposable.clear();
+        heartbeatDisposable.clear();
         listenersRegistered = false;
     }
 }

persistence-realm/src/main/java/chat/rocket/persistence/realm/repositories/RealmSessionRepository.java

@@ -7,6 +7,7 @@ import com.hadisatrio.optional.Optional;
 
 import chat.rocket.core.models.Session;
 import chat.rocket.core.repositories.SessionRepository;
+import chat.rocket.persistence.realm.RealmHelper;
 import chat.rocket.persistence.realm.RealmStore;
 import chat.rocket.persistence.realm.models.internal.RealmSession;
 import hu.akarnokd.rxjava.interop.RxJavaInterop;
@@ -35,7 +36,7 @@ public class RealmSessionRepository extends RealmRepository implements SessionRe
           return RxJavaInterop.toV2Flowable(pair.first.where(RealmSession.class)
                           .equalTo(RealmSession.ID, id)
                           .findAll()
-                          .<RealmSession>asObservable());
+                          .<RealmSession>asObservable().first());
         },
         pair -> close(pair.first, pair.second)
     )
@@ -74,14 +75,9 @@ public class RealmSessionRepository extends RealmRepository implements SessionRe
       realmSession.setTokenVerified(session.isTokenVerified());
       realmSession.setError(session.getError());
 
-      realm.beginTransaction();
-
-      return RxJavaInterop.toV2Flowable(realm.copyToRealmOrUpdate(realmSession)
-          .asObservable())
+      return RealmHelper.copyToRealmOrUpdate(realm, realmSession)
           .filter(it -> it != null && it.isLoaded() && it.isValid())
           .firstElement()
-          .doOnSuccess(it -> realm.commitTransaction())
-          .doOnError(throwable -> realm.cancelTransaction())
           .doOnEvent((realmObject, throwable) -> close(realm, looper))
           .toSingle()
           .map(realmObject -> true);