Merge branch 'develop' into iss321

android-ddp/src/main/java/chat/rocket/android_ddp/DDPClient.java

 package chat.rocket.android_ddp;
 
 import android.support.annotation.Nullable;
-import io.reactivex.Flowable;
+
 import org.json.JSONArray;
 
 import bolts.Task;
 import bolts.TaskCompletionSource;
 import chat.rocket.android_ddp.rx.RxWebSocketCallback;
+import io.reactivex.Flowable;
+import io.reactivex.Maybe;
 import okhttp3.OkHttpClient;
 
 public class DDPClient {
@@ -34,6 +36,10 @@ public class DDPClient {
     return task.getTask();
   }
 
+  public Maybe<DDPClientCallback.Base> doPing(@Nullable String id) {
+    return impl.ping(id);
+  }
+
   public Task<DDPClientCallback.RPC> rpc(String method, JSONArray params, String id,
                                          long timeoutMs) {
     TaskCompletionSource<DDPClientCallback.RPC> task = new TaskCompletionSource<>();

android-ddp/src/main/java/chat/rocket/android_ddp/DDPClientCallback.java

 package chat.rocket.android_ddp;
 
+import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 import org.json.JSONObject;
 
@@ -53,6 +54,15 @@ public class DDPClientCallback {
       this.id = id;
     }
 
+    public static class UnMatched extends Base {
+      @NonNull public String id;
+
+      public UnMatched(DDPClient client, @NonNull String id) {
+        super(client);
+        this.id = id;
+      }
+    }
+
     public static class Timeout extends BaseException {
       public Timeout(DDPClient client) {
         super(Timeout.class, client);

android-ddp/src/main/java/chat/rocket/android_ddp/DDPClientImpl.java

@@ -17,6 +17,7 @@ import chat.rocket.android.log.RCLog;
 import chat.rocket.android_ddp.rx.RxWebSocket;
 import chat.rocket.android_ddp.rx.RxWebSocketCallback;
 import io.reactivex.Flowable;
+import io.reactivex.Maybe;
 import io.reactivex.disposables.CompositeDisposable;
 import okhttp3.OkHttpClient;
 
@@ -106,40 +107,75 @@ public class DDPClientImpl {
     }
   }
 
-  public void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
-                   @Nullable final String id) {
+  public Maybe<DDPClientCallback.Base> ping(@Nullable final String id) {
 
     final boolean requested = (TextUtils.isEmpty(id)) ?
         sendMessage("ping", null) :
         sendMessage("ping", json -> json.put("id", id));
 
     if (requested) {
-      CompositeDisposable disposables = new CompositeDisposable();
-
-      disposables.add(
-          flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
-              .timeout(8, TimeUnit.SECONDS)
+      return flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
               .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
               .map(DDPClientImpl::toJson)
-              .subscribe(
-                  response -> {
-                    String msg = extractMsg(response);
-                    if ("pong".equals(msg)) {
-                      if (response.isNull("id")) {
-                        task.setResult(new DDPClientCallback.Ping(client, null));
-                        disposables.clear();
-                      } else {
-                        String _id = response.optString("id");
-                        if (id.equals(_id)) {
-                          task.setResult(new DDPClientCallback.Ping(client, id));
-                          disposables.clear();
-                        }
-                      }
-                      disposables.clear();
+              .filter(response -> "pong".equalsIgnoreCase(extractMsg(response)))
+              .doOnError(error -> {
+                RCLog.e(error, "Heartbeat ping[%s] xxx failed xxx", id);
+              })
+              .map(response -> {
+                String msg = extractMsg(response);
+                if ("pong".equals(msg)) {
+                  RCLog.d("pong[%s] <", id);
+                  if (response.isNull("id")) {
+                    return new DDPClientCallback.Ping(client, null);
+                  } else {
+                    String _id = response.optString("id");
+                    if (id.equals(_id)) {
+                      return new DDPClientCallback.Ping(client, _id);
+                    } else {
+                      return new DDPClientCallback.Ping.UnMatched(client, _id);
                     }
-                  },
-                  err -> task.setError(new DDPClientCallback.Ping.Timeout(client))
-              )
+                  }
+                }
+                // if we receive anything other than a pong throw an exception
+                throw new DDPClientCallback.RPC.Error(client, id, response);
+              }).firstElement();
+    } else {
+      return Maybe.error(new DDPClientCallback.Closed(client));
+    }
+  }
+
+  public void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
+                   @Nullable final String id) {
+
+    final boolean requested = (TextUtils.isEmpty(id)) ?
+            sendMessage("ping", null) :
+            sendMessage("ping", json -> json.put("id", id));
+
+    if (requested) {
+      CompositeDisposable disposables = new CompositeDisposable();
+
+      disposables.add(
+              flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
+                      .timeout(8, TimeUnit.SECONDS)
+                      .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
+                      .map(DDPClientImpl::toJson)
+                      .subscribe(
+                              response -> {
+                                String msg = extractMsg(response);
+                                if ("pong".equals(msg)) {
+                                  if (response.isNull("id")) {
+                                    task.setResult(new DDPClientCallback.Ping(client, null));
+                                  } else {
+                                    String _id = response.optString("id");
+                                    if (id.equals(_id)) {
+                                      task.setResult(new DDPClientCallback.Ping(client, id));
+                                    }
+                                  }
+                                  disposables.clear();
+                                }
+                              },
+                              err -> task.setError(new DDPClientCallback.Ping.Timeout(client))
+                      )
       );
 
       addErrorCallback(disposables, task);
@@ -368,12 +404,11 @@ public class DDPClientImpl {
     try {
       JSONObject origJson = new JSONObject().put("msg", msg);
       String msg2 = (json == null ? origJson : json.create(origJson)).toString();
-      websocket.sendText(msg2);
+      return websocket.sendText(msg2);
     } catch (Exception e) {
       RCLog.e(e);
       return false;
     }
-    return true; // ignore exception here.
   }
 
   private void sendMessage(String msg, @Nullable JSONBuilder json,
@@ -387,6 +422,9 @@ public class DDPClientImpl {
     disposables.add(
         flowable.subscribe(
             base -> {
+              if (base instanceof RxWebSocketCallback.Close) {
+                task.trySetError(new Exception(((RxWebSocketCallback.Close) base).reason));
+              }
             },
             err -> {
               task.trySetError(new Exception(err));

android-ddp/src/main/java/chat/rocket/android_ddp/rx/RxWebSocket.java

@@ -62,14 +62,20 @@ public class RxWebSocket {
               }
             }),
         BackpressureStrategy.BUFFER
-    ).delay(4, TimeUnit.SECONDS).publish();
+    ).publish();
   }
 
   public boolean sendText(String message) throws IOException {
+    if (webSocket == null) {
+        return false;
+    }
     return webSocket.send(message);
   }
 
   public boolean close(int code, String reason) throws IOException {
+    if (webSocket == null) {
+      return false;
+    }
     return webSocket.close(code, reason);
   }
 }

app/build.gradle

@@ -142,9 +142,15 @@ dependencies {
 
     compile "com.github.hotchemi:permissionsdispatcher:$permissionsdispatcherVersion"
     annotationProcessor "com.github.hotchemi:permissionsdispatcher-processor:$permissionsdispatcherVersion"
+
     compile('com.crashlytics.sdk.android:crashlytics:2.6.8@aar') {
         transitive = true;
     }
+
+    provided 'com.parse.bolts:bolts-tasks:1.4.0'
+    provided 'io.reactivex.rxjava2:rxjava:2.1.0'
+    provided 'io.reactivex:rxjava:1.3.0'
+    provided "com.github.akarnokd:rxjava2-interop:0.10.2"
 }
 
 apply plugin: 'com.google.gms.google-services'

app/src/main/AndroidManifest.xml

@@ -69,8 +69,18 @@
                 <action android:name="com.google.android.gms.iid.InstanceID"/>
             </intent-filter>
         </service>
+
+        <service android:name=".service.TaskService"
+            android:permission="com.google.android.gms.permission.BIND_NETWORK_TASK_SERVICE"
+            android:exported="true">
+            <intent-filter>
+                <action android:name="com.google.android.gms.gcm.ACTION_TASK_READY"/>
+            </intent-filter>
+        </service>
+
         <meta-data
             android:name="io.fabric.ApiKey"
             android:value="12ac6e94f850aaffcdff52001af77ca415d06a43" />
+
     </application>
 </manifest>
\ No newline at end of file

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/activity/MainActivity.java

@@ -34,7 +34,6 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
   private StatusTicker statusTicker;
 
   private MainContract.Presenter presenter;
-  private RoomFragment roomFragment;
 
   @Override
   protected int getLayoutContainerForFragment() {
@@ -180,8 +179,7 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
 
   @Override
   public void showRoom(String hostname, String roomId) {
-    roomFragment = RoomFragment.create(hostname, roomId);
-    showFragment(roomFragment);
+    showFragment(RoomFragment.create(hostname, roomId));
     closeSidebarIfNeeded();
     KeyboardHelper.hideSoftKeyboard(this);
   }
@@ -224,9 +222,6 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
   @Override
   public void showConnectionOk() {
     statusTicker.updateStatus(StatusTicker.STATUS_DISMISS, null);
-    if (roomFragment != null) {
-      roomFragment.refreshRoom();
-    }
   }
 
   //TODO: consider this class to define in layouthelper for more complicated operation.

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

@@ -98,8 +98,10 @@ public class MainPresenter extends BasePresenter<MainContract.View>
 
   @Override
   public void onRetryLogin() {
-    view.showConnecting();
-    connectivityManagerApi.keepAliveServer();
+    final Disposable subscription = sessionInteractor.retryLogin()
+            .subscribe();
+
+    addSubscription(subscription);
   }
 
   private void openRoom() {

app/src/main/java/chat/rocket/android/api/DDPClientWrapper.java

@@ -15,6 +15,7 @@ import chat.rocket.android_ddp.DDPClient;
 import chat.rocket.android_ddp.DDPClientCallback;
 import chat.rocket.android_ddp.DDPSubscription;
 import io.reactivex.Flowable;
+import io.reactivex.Maybe;
 
 /**
  * DDP client wrapper.
@@ -124,4 +125,13 @@ public class DDPClientWrapper {
           }
         });
   }
+
+  /**
+   * check WebSocket connectivity with ping.
+   */
+  public Maybe<DDPClientCallback.Base> doPing() {
+    final String pingId = UUID.randomUUID().toString();
+    RCLog.d("ping[%s] >", pingId);
+    return ddpClient.doPing(pingId);
+  }
 }

app/src/main/java/chat/rocket/android/api/MethodCallHelper.java

@@ -83,6 +83,10 @@ public class MethodCallHelper {
     return task.continueWithTask(_task -> {
       if (_task.isFaulted()) {
         Exception exception = _task.getError();
+        // If wet get any error, close the socket to let the RocketChatWebSocketThread aware of it.
+        // FIXME: when rewriting the network layer we should get rid of this MethodCallHelper
+        // monolith concept. It decouples a lot the socket from the rest of the app.
+        ddpClientRef.get().close();
         if (exception instanceof MethodCall.Error) {
           String errMessageJson = exception.getMessage();
           if (TextUtils.isEmpty(errMessageJson)) {
@@ -94,7 +98,6 @@ public class MethodCallHelper {
           if (TwoStepAuthException.TYPE.equals(errType)) {
             return Task.forError(new TwoStepAuthException(errMessage));
           }
-
           return Task.forError(new Exception(errMessage));
         } else if (exception instanceof DDPClientCallback.RPC.Error) {
           String errMessage = ((DDPClientCallback.RPC.Error) exception).error.getString("message");

app/src/main/java/chat/rocket/android/fragment/chatroom/RoomFragment.java

@@ -602,8 +602,4 @@ public class RoomFragment extends AbstractChatRoomFragment implements
     edittingMessage = message;
     messageFormManager.setEditMessage(message.getMessage());
   }
-
-  public void refreshRoom() {
-    presenter.loadMessages();
-  }
 }
\ No newline at end of file

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

@@ -63,6 +63,7 @@ import rx.subjects.PublishSubject;
     }
   }
 
+  @DebugLog
   @Override
   public void ensureConnections() {
     for (String hostname : serverConnectivityList.keySet()) {
@@ -146,6 +147,7 @@ import rx.subjects.PublishSubject;
     return Observable.concat(Observable.from(getCurrentConnectivityList()), connectivitySubject);
   }
 
+  @DebugLog
   private Single<Boolean> connectToServerIfNeeded(String hostname, boolean forceConnect) {
     return Single.defer(() -> {
       final int connectivity = serverConnectivityList.get(hostname);
@@ -163,8 +165,8 @@ import rx.subjects.PublishSubject;
       }
 
       return connectToServer(hostname)
-          //.doOnError(RCLog::e)
-          .retryWhen(RxHelper.exponentialBackoff(3, 500, TimeUnit.MILLISECONDS));
+          .doOnError(RCLog::e)
+          .retryWhen(RxHelper.exponentialBackoff(Integer.MAX_VALUE, 500, TimeUnit.MILLISECONDS));
     });
   }
 
@@ -191,7 +193,7 @@ import rx.subjects.PublishSubject;
     });
   }
 
-
+  @DebugLog
   private Single<Boolean> waitForConnected(String hostname) {
     return connectivitySubject
         .filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))
@@ -207,6 +209,7 @@ import rx.subjects.PublishSubject;
                 : Single.error(new ServerConnectivity.DisconnectedException()));
   }
 
+  @DebugLog
   private Single<Boolean> waitForDisconnected(String hostname) {
     return connectivitySubject
         .filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))

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

@@ -4,15 +4,18 @@ import android.content.Context;
 import android.os.Handler;
 import android.os.HandlerThread;
 
+import com.google.android.gms.gcm.GcmNetworkManager;
+import com.google.android.gms.gcm.PeriodicTask;
+
 import org.json.JSONObject;
 
 import java.lang.reflect.Constructor;
 import java.util.ArrayList;
 import java.util.Iterator;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import bolts.Task;
-import chat.rocket.android.RocketChatCache;
 import chat.rocket.android.api.DDPClientWrapper;
 import chat.rocket.android.api.MethodCallHelper;
 import chat.rocket.android.helper.LogIfError;
@@ -22,7 +25,6 @@ import chat.rocket.android.log.RCLog;
 import chat.rocket.android.service.ddp.base.ActiveUsersSubscriber;
 import chat.rocket.android.service.ddp.base.LoginServiceConfigurationSubscriber;
 import chat.rocket.android.service.ddp.base.UserDataSubscriber;
-import chat.rocket.android.service.ddp.stream.StreamRoomMessage;
 import chat.rocket.android.service.observer.CurrentUserObserver;
 import chat.rocket.android.service.observer.FileUploadingToUrlObserver;
 import chat.rocket.android.service.observer.FileUploadingWithUfsObserver;
@@ -33,12 +35,15 @@ import chat.rocket.android.service.observer.MethodCallObserver;
 import chat.rocket.android.service.observer.NewMessageObserver;
 import chat.rocket.android.service.observer.PushSettingsObserver;
 import chat.rocket.android.service.observer.SessionObserver;
-import chat.rocket.android.service.observer.TokenLoginObserver;
+import chat.rocket.android_ddp.DDPClientCallback;
 import chat.rocket.core.models.ServerInfo;
 import chat.rocket.persistence.realm.RealmHelper;
 import chat.rocket.persistence.realm.RealmStore;
 import chat.rocket.persistence.realm.models.internal.RealmSession;
 import hugo.weaving.DebugLog;
+import io.reactivex.Flowable;
+import io.reactivex.disposables.CompositeDisposable;
+import rx.Completable;
 import rx.Single;
 import rx.subscriptions.CompositeSubscription;
 
@@ -50,7 +55,6 @@ public class RocketChatWebSocketThread extends HandlerThread {
       LoginServiceConfigurationSubscriber.class,
       ActiveUsersSubscriber.class,
       UserDataSubscriber.class,
-      TokenLoginObserver.class,
       MethodCallObserver.class,
       SessionObserver.class,
       LoadMessageProcedureObserver.class,
@@ -62,14 +66,16 @@ public class RocketChatWebSocketThread extends HandlerThread {
       PushSettingsObserver.class,
       GcmPushRegistrationObserver.class
   };
+  private static final long HEARTBEAT_PERIOD_MS = 20000;
   private final Context appContext;
   private final String hostname;
   private final RealmHelper realmHelper;
   private final ConnectivityManagerInternal connectivityManager;
   private final ArrayList<Registrable> listeners = new ArrayList<>();
+  private final CompositeDisposable hearbeatDisposable = new CompositeDisposable();
+  private final CompositeSubscription reconnectSubscription = new CompositeSubscription();
   private DDPClientWrapper ddpClient;
   private boolean listenersRegistered;
-  private RocketChatCache rocketChatCache;
   private final DDPClientRef ddpClientRef = new DDPClientRef() {
     @Override
     public DDPClientWrapper get() {
@@ -103,7 +109,6 @@ public class RocketChatWebSocketThread extends HandlerThread {
     this.hostname = hostname;
     this.realmHelper = RealmStore.getOrCreate(hostname);
     this.connectivityManager = ConnectivityManager.getInstanceForInternal(appContext);
-    this.rocketChatCache = new RocketChatCache(appContext);
   }
 
   /**
@@ -185,9 +190,10 @@ public class RocketChatWebSocketThread extends HandlerThread {
   @DebugLog
   public Single<Boolean> keepAlive() {
     return checkIfConnectionAlive()
-        .flatMap(alive -> alive ? Single.just(true) : connect());
+        .flatMap(alive -> alive ? Single.just(true) : connectWithExponentialBackoff());
   }
 
+  @DebugLog
   private Single<Boolean> checkIfConnectionAlive() {
     if (ddpClient == null) {
       return Single.just(false);
@@ -206,7 +212,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
             if (task.isFaulted()) {
               Exception error = task.getError();
               RCLog.e(error);
-              emitter.onSuccess(false);
+              emitter.onError(error);
             } else {
               keepAliveTimer.update();
               emitter.onSuccess(true);
@@ -218,8 +224,28 @@ public class RocketChatWebSocketThread extends HandlerThread {
     });
   }
 
+  @DebugLog
+  private Flowable<Boolean> heartbeat(long interval) {
+    return Flowable.interval(interval, TimeUnit.MILLISECONDS)
+            .onBackpressureDrop()
+            .flatMap(tick -> ddpClient.doPing().toFlowable())
+            .map(callback -> {
+              if (callback instanceof DDPClientCallback.Ping) {
+                return true;
+              }
+              // ideally we should never get here. We should always receive a DDPClientCallback.Ping
+              // because we just received a pong. But maybe we received a pong from an unmatched
+              // ping id which we should ignore. In this case or any other random error, log and
+              // send false downstream
+              RCLog.d("heartbeat pong < %s", callback.toString());
+              return false;
+            });
+  }
+
   private Single<Boolean> prepareDDPClient() {
-    return checkIfConnectionAlive()
+    // TODO: temporarily replaced checkIfConnectionAlive() call for this single checking if ddpClient is
+    // null or not. In case it is, create a new client, otherwise just keep connecting with existing one.
+    return Single.just(ddpClient != null)
         .doOnSuccess(alive -> {
           if (!alive) {
             RCLog.d("DDPClient#create");
@@ -240,26 +266,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
 
                 // handling WebSocket#onClose() callback.
                 task.getResult().client.getOnCloseCallback().onSuccess(_task -> {
-                  ddpClient.close();
-                  forceInvalidateTokens();
-                  connectivityManager.notifyConnecting(hostname);
-                  // Needed to use subscriptions because of legacy code.
-                  // TODO: Should update to RxJava 2
-                  final CompositeSubscription subscriptions = new CompositeSubscription();
-                  subscriptions.add(
-                    connect().retryWhen(RxHelper.exponentialBackoff(3, 500, TimeUnit.MILLISECONDS))
-                    .subscribe(
-                        connected -> {
-                          if (!connected) {
-                            connectivityManager.notifyConnectionLost(
-                                    hostname, ConnectivityManagerInternal.REASON_NETWORK_ERROR
-                            );
-                          }
-                          subscriptions.clear();
-                        },
-                        err -> logErrorAndUnsubscribe(subscriptions, err)
-                    )
-                  );
+                  reconnect();
                   return null;
                 });
 
@@ -290,11 +297,39 @@ public class RocketChatWebSocketThread extends HandlerThread {
         }));
   }
 
+  private void reconnect() {
+    // if we are already trying to reconnect then return.
+    if (reconnectSubscription.hasSubscriptions()) {
+      return;
+    }
+    ddpClient.close();
+    forceInvalidateTokens();
+    connectivityManager.notifyConnecting(hostname);
+    // Needed to use subscriptions because of legacy code.
+    // TODO: Should update to RxJava 2
+    reconnectSubscription.add(
+            connectWithExponentialBackoff()
+                    .subscribe(
+                            connected -> {
+                              if (!connected) {
+                                connectivityManager.notifyConnecting(hostname);
+                              }
+                              reconnectSubscription.clear();
+                            },
+                            err -> logErrorAndUnsubscribe(reconnectSubscription, err)
+                    )
+    );
+  }
+
   private void logErrorAndUnsubscribe(CompositeSubscription subscriptions, Throwable err) {
     RCLog.e(err);
     subscriptions.clear();
   }
 
+  private Single<Boolean> connectWithExponentialBackoff() {
+    return connect().retryWhen(RxHelper.exponentialBackoff(Integer.MAX_VALUE, 500, TimeUnit.MILLISECONDS));
+  }
+
   @DebugLog
   private Single<Boolean> connect() {
     return connectDDPClient()
@@ -325,12 +360,48 @@ public class RocketChatWebSocketThread extends HandlerThread {
     if (listenersRegistered) {
       unregisterListeners();
     }
-    listenersRegistered = true;
 
+    List<RealmSession> sessions = realmHelper.executeTransactionForReadResults(realm ->
+            realm.where(RealmSession.class)
+                    .isNotNull(RealmSession.TOKEN)
+                    .equalTo(RealmSession.TOKEN_VERIFIED, false)
+                    .isNull(RealmSession.ERROR)
+                    .findAll());
+
+    if (sessions != null && sessions.size() > 0) {
+      // if we have a session try to resume it. At this point we're probably recovering from
+      // a disconnection state
+      final CompositeSubscription subscriptions = new CompositeSubscription();
+      MethodCallHelper methodCall = new MethodCallHelper(realmHelper, ddpClientRef);
+      subscriptions.add(
+              Completable.defer(() -> {
+                Task<Void> result = methodCall.loginWithToken(sessions.get(0).getToken());
+                if (result.isFaulted()) {
+                  return Completable.error(result.getError());
+                } else {
+                  return Completable.complete();
+                }
+              }).retryWhen(RxHelper.exponentialBackoff(Integer.MAX_VALUE, 500, TimeUnit.MILLISECONDS))
+                .subscribe(
+                      () -> {
+                        createObserversAndRegister();
+                        subscriptions.clear();
+                      },
+                      error -> logErrorAndUnsubscribe(subscriptions, error)
+              )
+      );
+    } else {
+      // if we don't have any session then just create the observers and register normally
+      createObserversAndRegister();
+    }
+  }
+
+  @DebugLog
+  private void createObserversAndRegister() {
     for (Class clazz : REGISTERABLE_CLASSES) {
       try {
         Constructor ctor = clazz.getConstructor(Context.class, String.class, RealmHelper.class,
-            DDPClientRef.class);
+                DDPClientRef.class);
         Object obj = ctor.newInstance(appContext, hostname, realmHelper, ddpClientRef);
 
         if (obj instanceof Registrable) {
@@ -338,19 +409,57 @@ public class RocketChatWebSocketThread extends HandlerThread {
           registrable.register();
           listeners.add(registrable);
         }
-        // Register for room stream messages
-        String roomId = rocketChatCache.getSelectedRoomId();
-        if (roomId != null && !roomId.isEmpty()) {
-          StreamRoomMessage streamRoomMessage = new StreamRoomMessage(
-                  appContext, hostname, realmHelper, ddpClientRef, roomId
-          );
-          streamRoomMessage.register();
-          listeners.add(streamRoomMessage);
-        }
       } catch (Exception exception) {
         RCLog.w(exception, "Failed to register listeners!!");
       }
     }
+    listenersRegistered = true;
+    startHeartBeat();
+  }
+
+  private void startHeartBeat() {
+    // This task is scheduled to guarantee that RocketChatService is still bound at the application
+    // process. This is necessary due to the way the keep-alive assertions are currently architectured.
+    // By doing those keep-alives periodically we try to ensure that we have the app alive
+    // if for some reason its process gets killed (for any reason).
+    // TODO: should set this at another point; we should specify a reasonable time-window.
+    // TODO: should check and handle the case Google Play Services isn't available.
+    // TODO: consider on using https://github.com/evernote/android-job for this as it allows much more
+    // customisation like running at exponential backoff and others. We should alos use it more
+    // extensively throughout the app on all the common tasks, like i.e. sending a message
+    GcmNetworkManager gcmNetworkManager = GcmNetworkManager.getInstance(appContext);
+    gcmNetworkManager.schedule(
+            new PeriodicTask.Builder()
+                    .setRequiresCharging(false)
+                    .setUpdateCurrent(true)
+                    .setRequiredNetwork(com.google.android.gms.gcm.Task.NETWORK_STATE_ANY)
+                    .setTag(TaskService.TAG_KEEP_ALIVE)
+                    .setService(TaskService.class)
+                    .setPeriod(30)
+                    .setFlex(15)
+                    .build()
+    );
+
+    hearbeatDisposable.clear();
+    hearbeatDisposable.add(
+        heartbeat(HEARTBEAT_PERIOD_MS)
+            .subscribe(
+                    ponged -> {
+                      if (!ponged) {
+                        RCLog.d("Pong received but didn't match ping id");
+                      }
+                    },
+                    error -> {
+                      RCLog.e(error);
+                      // Stop pinging
+                      hearbeatDisposable.clear();
+                      if (error instanceof DDPClientCallback.Closed) {
+                        RCLog.d("Hearbeat failure: retrying connection...");
+                        reconnect();
+                      }
+                    }
+            )
+    );
   }
 
   @DebugLog
@@ -370,6 +479,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
       registrable.unregister();
       iterator.remove();
     }
+    hearbeatDisposable.clear();
     listenersRegistered = false;
   }
 }

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

+package chat.rocket.android.service;
+
+import com.google.android.gms.gcm.GcmNetworkManager;
+import com.google.android.gms.gcm.GcmTaskService;
+import com.google.android.gms.gcm.TaskParams;
+
+public class TaskService extends GcmTaskService {
+
+    public static final String TAG_KEEP_ALIVE = "TAG_KEEP_ALIVE";
+
+    @Override
+    public int onRunTask(TaskParams taskParams) {
+        switch (taskParams.getTag()) {
+            case TAG_KEEP_ALIVE:
+                ConnectivityManager.getInstance(getApplicationContext()).keepAliveServer();
+                return GcmNetworkManager.RESULT_SUCCESS;
+            default:
+                return GcmNetworkManager.RESULT_FAILURE;
+        }
+    }
+}

app/src/main/java/chat/rocket/android/service/ddp/base/AbstractBaseSubscriber.java

@@ -20,7 +20,7 @@ abstract class AbstractBaseSubscriber extends AbstractDDPDocEventSubscriber {
 
   @Override
   protected final boolean shouldTruncateTableOnInitialize() {
-    return true;
+    return false;
   }
 
   protected abstract String getSubscriptionCallbackName();

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

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

@@ -4,6 +4,7 @@ import android.content.Context;
 import android.os.Handler;
 import android.os.Looper;
 
+import chat.rocket.android.RocketChatCache;
 import chat.rocket.persistence.realm.RealmHelper;
 import chat.rocket.android.service.DDPClientRef;
 import chat.rocket.android.service.Registrable;
@@ -19,6 +20,7 @@ public class StreamRoomMessageManager implements Registrable {
   private final DDPClientRef ddpClientRef;
   private final AbstractRocketChatCacheObserver cacheObserver;
   private final Handler handler;
+  private final RocketChatCache rocketChatCache;
   private StreamRoomMessage streamRoomMessage;
 
   public StreamRoomMessageManager(Context context, String hostname,
@@ -27,6 +29,7 @@ public class StreamRoomMessageManager implements Registrable {
     this.hostname = hostname;
     this.realmHelper = realmHelper;
     this.ddpClientRef = ddpClientRef;
+    this.rocketChatCache = new RocketChatCache(context);
 
     cacheObserver = new AbstractRocketChatCacheObserver(context, realmHelper) {
       @Override
@@ -57,6 +60,11 @@ public class StreamRoomMessageManager implements Registrable {
   @Override
   public void register() {
     cacheObserver.register();
+    String selectedRoomId = rocketChatCache.getSelectedRoomId();
+    if (selectedRoomId == null) {
+      return;
+    }
+    registerStreamNotifyMessage(selectedRoomId);
   }
 
   @Override

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

@@ -135,13 +135,15 @@ public class RealmMessageRepository extends RealmRepository implements MessageRe
         () -> new Pair<>(RealmStore.getRealm(hostname), Looper.myLooper()),
         pair -> RxJavaInterop.toV2Flowable(pair.first.where(RealmMessage.class)
             .equalTo(RealmMessage.ROOM_ID, room.getRoomId())
+            .isNotNull(RealmMessage.USER)
             .findAllSorted(RealmMessage.TIMESTAMP, Sort.DESCENDING)
             .asObservable()),
         pair -> close(pair.first, pair.second)
     )
         .unsubscribeOn(AndroidSchedulers.from(Looper.myLooper()))
         .filter(it -> it.isLoaded() && it.isValid())
-        .map(this::toList));
+        .map(this::toList)
+        .distinctUntilChanged());
   }
 
   @Override