Merge pull request #354 from RocketChat/fix/websocket-drops-keeping-bolts

Fix/websocket drops keeping bolts

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

 package chat.rocket.android_ddp.rx;
 
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import chat.rocket.android.log.RCLog;
 import io.reactivex.BackpressureStrategy;
 import io.reactivex.Flowable;
 import io.reactivex.FlowableOnSubscribe;
 import io.reactivex.exceptions.OnErrorNotImplementedException;
 import io.reactivex.flowables.ConnectableFlowable;
-
-import java.io.IOException;
-import chat.rocket.android.log.RCLog;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
@@ -15,8 +16,10 @@ import okhttp3.WebSocket;
 import okhttp3.WebSocketListener;
 
 public class RxWebSocket {
+  public static final int REASON_NETWORK_ERROR = 100;
   private OkHttpClient httpClient;
   private WebSocket webSocket;
+  private boolean hadErrorsBefore;
 
   public RxWebSocket(OkHttpClient client) {
     httpClient = client;
@@ -30,6 +33,7 @@ public class RxWebSocket {
             .newWebSocket(request, new WebSocketListener() {
               @Override
               public void onOpen(WebSocket webSocket, Response response) {
+                hadErrorsBefore = false;
                 RxWebSocket.this.webSocket = webSocket;
                 emitter.onNext(new RxWebSocketCallback.Open(RxWebSocket.this.webSocket, response));
               }
@@ -37,7 +41,11 @@ public class RxWebSocket {
               @Override
               public void onFailure(WebSocket webSocket, Throwable err, Response response) {
                 try {
-                  emitter.onError(new RxWebSocketCallback.Failure(webSocket, err, response));
+                  if (!hadErrorsBefore) {
+                    hadErrorsBefore = true;
+                    emitter.onNext(new RxWebSocketCallback.Close(webSocket, REASON_NETWORK_ERROR, err.getMessage()));
+                    emitter.onComplete();
+                  }
                 } catch (OnErrorNotImplementedException ex) {
                   RCLog.w(ex, "OnErrorNotImplementedException ignored");
                 }
@@ -51,11 +59,10 @@ public class RxWebSocket {
               @Override
               public void onClosed(WebSocket webSocket, int code, String reason) {
                 emitter.onNext(new RxWebSocketCallback.Close(webSocket, code, reason));
-                emitter.onComplete();
               }
             }),
         BackpressureStrategy.BUFFER
-    ).publish();
+    ).delay(4, TimeUnit.SECONDS).publish();
   }
 
   public boolean sendText(String message) throws IOException {

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

 package chat.rocket.android_ddp.rx;
 
-import static android.R.attr.type;
-
 import chat.rocket.android.log.RCLog;
 import okhttp3.Response;
 import okhttp3.WebSocket;
@@ -28,25 +26,8 @@ public class RxWebSocketCallback {
     public Open(WebSocket websocket, Response response) {
       super("Open", websocket);
       this.response = response;
-    }
-  }
-
-  public static class Failure extends Exception {
-    public WebSocket ws;
-    public Response response;
-
-    public Failure(WebSocket websocket, Throwable err, Response response) {
-      super(err);
-      this.ws = websocket;
-      this.response = response;
-    }
-
-    @Override
-    public String toString() {
-      if (response != null) {
-        return "[" + type + "] " + response.message();
-      } else {
-        return super.toString();
+      if (response != null && response.body() != null) {
+        this.response.body().close();
       }
     }
   }

app/build.gradle

@@ -35,7 +35,7 @@ android {
         applicationId "chat.rocket.android"
         minSdkVersion 16
         targetSdkVersion rootProject.ext.targetSdkVersion
-        versionCode 26
+        versionCode 27
         versionName "1.0.16"
         testInstrumentationRunner "android.support.test.runner.AndroidJUnitRunner"
 

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

@@ -16,11 +16,11 @@ import chat.rocket.android.fragment.chatroom.HomeFragment;
 import chat.rocket.android.fragment.chatroom.RoomFragment;
 import chat.rocket.android.fragment.sidebar.SidebarMainFragment;
 import chat.rocket.android.helper.KeyboardHelper;
+import chat.rocket.android.service.ConnectivityManager;
+import chat.rocket.android.widget.RoomToolbar;
 import chat.rocket.core.interactors.CanCreateRoomInteractor;
 import chat.rocket.core.interactors.RoomInteractor;
 import chat.rocket.core.interactors.SessionInteractor;
-import chat.rocket.android.service.ConnectivityManager;
-import chat.rocket.android.widget.RoomToolbar;
 import chat.rocket.persistence.realm.repositories.RealmRoomRepository;
 import chat.rocket.persistence.realm.repositories.RealmSessionRepository;
 import chat.rocket.persistence.realm.repositories.RealmUserRepository;
@@ -34,6 +34,7 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
   private StatusTicker statusTicker;
 
   private MainContract.Presenter presenter;
+  private RoomFragment roomFragment;
 
   @Override
   protected int getLayoutContainerForFragment() {
@@ -179,7 +180,8 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
 
   @Override
   public void showRoom(String hostname, String roomId) {
-    showFragment(RoomFragment.create(hostname, roomId));
+    roomFragment = RoomFragment.create(hostname, roomId);
+    showFragment(roomFragment);
     closeSidebarIfNeeded();
     KeyboardHelper.hideSoftKeyboard(this);
   }
@@ -222,6 +224,9 @@ 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

@@ -3,22 +3,23 @@ package chat.rocket.android.activity;
 import android.support.annotation.NonNull;
 import android.support.v4.util.Pair;
 
-import io.reactivex.Flowable;
-import io.reactivex.android.schedulers.AndroidSchedulers;
-import io.reactivex.disposables.Disposable;
-
 import chat.rocket.android.BackgroundLooper;
 import chat.rocket.android.RocketChatCache;
 import chat.rocket.android.api.MethodCallHelper;
 import chat.rocket.android.helper.LogIfError;
 import chat.rocket.android.helper.Logger;
 import chat.rocket.android.service.ConnectivityManagerApi;
+import chat.rocket.android.service.ServerConnectivity;
 import chat.rocket.android.shared.BasePresenter;
 import chat.rocket.core.interactors.CanCreateRoomInteractor;
 import chat.rocket.core.interactors.RoomInteractor;
 import chat.rocket.core.interactors.SessionInteractor;
 import chat.rocket.core.models.Session;
 import chat.rocket.core.models.User;
+import hu.akarnokd.rxjava.interop.RxJavaInterop;
+import io.reactivex.Flowable;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.disposables.Disposable;
 
 public class MainPresenter extends BasePresenter<MainContract.View>
     implements MainContract.Presenter {
@@ -63,6 +64,7 @@ public class MainPresenter extends BasePresenter<MainContract.View>
 
     openRoom();
 
+    subscribeToNetworkChanges();
     subscribeToUnreadCount();
     subscribeToSession();
     setUserOnline();
@@ -96,10 +98,8 @@ public class MainPresenter extends BasePresenter<MainContract.View>
 
   @Override
   public void onRetryLogin() {
-    final Disposable subscription = sessionInteractor.retryLogin()
-        .subscribe();
-
-    addSubscription(subscription);
+    view.showConnecting();
+    connectivityManagerApi.keepAliveServer();
   }
 
   private void openRoom() {
@@ -161,6 +161,23 @@ public class MainPresenter extends BasePresenter<MainContract.View>
     addSubscription(subscription);
   }
 
+  private void subscribeToNetworkChanges() {
+    Disposable disposable = RxJavaInterop.toV2Flowable(connectivityManagerApi.getServerConnectivityAsObservable())
+            .observeOn(AndroidSchedulers.mainThread())
+            .subscribe(
+                    connectivity -> {
+                      if (connectivity.state == ServerConnectivity.STATE_CONNECTED) {
+                        view.showConnectionOk();
+                        return;
+                      }
+                      view.showConnecting();
+                    },
+                    Logger::report
+            );
+
+    addSubscription(disposable);
+  }
+
   private void setUserOnline() {
     methodCallHelper.setUserPresence(User.STATUS_ONLINE)
         .continueWith(new LogIfError());

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

 package chat.rocket.android.api;
 
 import android.support.annotation.Nullable;
-import io.reactivex.Flowable;
+
 import org.json.JSONArray;
 import org.json.JSONException;
 
 import java.util.UUID;
+
 import bolts.Task;
 import chat.rocket.android.helper.OkHttpHelper;
 import chat.rocket.android.helper.TextUtils;
@@ -13,6 +14,7 @@ import chat.rocket.android.log.RCLog;
 import chat.rocket.android_ddp.DDPClient;
 import chat.rocket.android_ddp.DDPClientCallback;
 import chat.rocket.android_ddp.DDPSubscription;
+import io.reactivex.Flowable;
 
 /**
  * DDP client wrapper.

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

@@ -49,5 +49,7 @@ public interface RoomContract {
     void onUnreadCount();
 
     void onMarkAsRead();
+
+    void refreshRoom();
   }
 }

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

@@ -21,6 +21,14 @@ import android.support.v7.widget.LinearLayoutManager;
 import android.support.v7.widget.RecyclerView;
 import android.view.View;
 import android.view.ViewGroup;
+
+import com.fernandocejas.arrow.optional.Optional;
+import com.jakewharton.rxbinding2.support.v4.widget.RxDrawerLayout;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+
 import chat.rocket.android.BackgroundLooper;
 import chat.rocket.android.R;
 import chat.rocket.android.api.MethodCallHelper;
@@ -70,15 +78,10 @@ import chat.rocket.persistence.realm.repositories.RealmSessionRepository;
 import chat.rocket.persistence.realm.repositories.RealmSpotlightRoomRepository;
 import chat.rocket.persistence.realm.repositories.RealmSpotlightUserRepository;
 import chat.rocket.persistence.realm.repositories.RealmUserRepository;
-import com.fernandocejas.arrow.optional.Optional;
-import com.jakewharton.rxbinding2.support.v4.widget.RxDrawerLayout;
 import io.reactivex.Single;
 import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.disposables.CompositeDisposable;
 import io.reactivex.disposables.Disposable;
-import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.List;
 import permissions.dispatcher.NeedsPermission;
 import permissions.dispatcher.RuntimePermissions;
 
@@ -251,8 +254,8 @@ public class RoomFragment extends AbstractChatRoomFragment implements
 
   protected Snackbar getUnreadCountIndicatorView(int count) {
     // TODO: replace with another custom View widget, not to hide message composer.
-    final String caption = getResources().getString(
-        R.string.fmt_dialog_view_latest_message_title, count);
+    final String caption = getResources().getQuantityString(
+        R.plurals.fmt_dialog_view_latest_message_title, count, count);
 
     return Snackbar.make(rootView, caption, Snackbar.LENGTH_LONG)
         .setAction(R.string.dialog_view_latest_message_action, view -> scrollToLatestMessage());
@@ -599,4 +602,8 @@ 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/fragment/chatroom/RoomPresenter.java

@@ -55,7 +55,11 @@ public class RoomPresenter extends BasePresenter<RoomContract.View>
   @Override
   public void bindView(@NonNull RoomContract.View view) {
     super.bindView(view);
+    refreshRoom();
+  }
 
+  @Override
+  public void refreshRoom() {
     getRoomRoles();
     getRoomInfo();
     getRoomHistoryStateInfo();

app/src/main/java/chat/rocket/android/fragment/chatroom/dialog/UsersOfRoomDialogFragment.java

@@ -192,7 +192,7 @@ public class UsersOfRoomDialogFragment extends AbstractChatRoomDialogFragment {
    */
   private void onRenderTotalCount(long total) {
     TextView userCount = (TextView) getDialog().findViewById(R.id.room_user_count);
-    userCount.setText(getString(R.string.fmt_room_user_count, total));
+    userCount.setText(getResources().getQuantityString(R.plurals.fmt_room_user_count, (int) total, total));
   }
 
   /**

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

@@ -23,4 +23,6 @@ import chat.rocket.core.models.ServerInfo;
   void notifyConnectionEstablished(String hostname, String session);
 
   void notifyConnectionLost(String hostname, int reason);
+
+  void notifyConnecting(String hostname);
 }

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

@@ -133,6 +133,14 @@ import rx.subjects.PublishSubject;
         new ServerConnectivity(hostname, ServerConnectivity.STATE_DISCONNECTED));
   }
 
+  @DebugLog
+  @Override
+  public void notifyConnecting(String hostname) {
+    serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTING);
+    connectivitySubject.onNext(
+            new ServerConnectivity(hostname, ServerConnectivity.STATE_CONNECTING));
+  }
+
   @Override
   public Observable<ServerConnectivity> getServerConnectivityAsObservable() {
     return Observable.concat(Observable.from(getCurrentConnectivityList()), connectivitySubject);

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

@@ -9,6 +9,7 @@ import android.os.IBinder;
 import android.support.annotation.Nullable;
 
 import java.util.HashMap;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
 import chat.rocket.android.activity.MainActivity;
@@ -24,6 +25,7 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
 
   private ConnectivityManagerInternal connectivityManager;
   private HashMap<String, RocketChatWebSocketThread> webSocketThreads;
+  private Semaphore webSocketThreadLock = new Semaphore(1);
 
   public class LocalBinder extends Binder {
     ConnectivityServiceInterface getServiceInterface() {
@@ -49,6 +51,7 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
     context.unbindService(serviceConnection);
   }
 
+  @DebugLog
   @Override
   public void onCreate() {
     super.onCreate();
@@ -57,6 +60,7 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
     webSocketThreads = new HashMap<>();
   }
 
+  @DebugLog
   @Override
   public int onStartCommand(Intent intent, int flags, int startId) {
     connectivityManager.ensureConnections();
@@ -106,12 +110,17 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
   @DebugLog
   private Single<RocketChatWebSocketThread> getOrCreateWebSocketThread(String hostname) {
     return Single.defer(() -> {
+      webSocketThreadLock.acquire();
       if (webSocketThreads.containsKey(hostname)) {
         RocketChatWebSocketThread thread = webSocketThreads.get(hostname);
+        webSocketThreadLock.release();
         return Single.just(thread);
       }
       return RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
-          .doOnSuccess(thread -> webSocketThreads.put(hostname, thread));
+          .doOnSuccess(thread -> {
+            webSocketThreads.put(hostname, thread);
+            webSocketThreadLock.release();
+          });
     });
   }
 

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

@@ -3,24 +3,26 @@ package chat.rocket.android.service;
 import android.content.Context;
 import android.os.Handler;
 import android.os.HandlerThread;
+
 import org.json.JSONObject;
 
 import java.lang.reflect.Constructor;
 import java.util.ArrayList;
 import java.util.Iterator;
+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;
+import chat.rocket.android.helper.RxHelper;
 import chat.rocket.android.helper.TextUtils;
 import chat.rocket.android.log.RCLog;
-import chat.rocket.core.models.ServerInfo;
-import chat.rocket.persistence.realm.models.internal.RealmSession;
-import chat.rocket.persistence.realm.RealmHelper;
-import chat.rocket.persistence.realm.RealmStore;
 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;
@@ -32,8 +34,13 @@ 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.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 rx.Single;
+import rx.subscriptions.CompositeSubscription;
 
 /**
  * Thread for handling WebSocket connection.
@@ -62,6 +69,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
   private final ArrayList<Registrable> listeners = new ArrayList<>();
   private DDPClientWrapper ddpClient;
   private boolean listenersRegistered;
+  private RocketChatCache rocketChatCache;
   private final DDPClientRef ddpClientRef = new DDPClientRef() {
     @Override
     public DDPClientWrapper get() {
@@ -69,7 +77,6 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
   };
 
-
   private static class KeepAliveTimer {
     private long lastTime;
     private final long thresholdMs;
@@ -96,6 +103,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
     this.hostname = hostname;
     this.realmHelper = RealmStore.getOrCreate(hostname);
     this.connectivityManager = ConnectivityManager.getInstanceForInternal(appContext);
+    this.rocketChatCache = new RocketChatCache(appContext);
   }
 
   /**
@@ -147,7 +155,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
       return Single.fromEmitter(emitter -> {
         new Handler(getLooper()).post(() -> {
           RCLog.d("thread %s: terminated()", Thread.currentThread().getId());
-          unregisterListeners();
+          unregisterListenersAndClose();
           connectivityManager.notifyConnectionLost(hostname,
               ConnectivityManagerInternal.REASON_CLOSED_BY_USER);
           RocketChatWebSocketThread.super.quit();
@@ -196,9 +204,9 @@ public class RocketChatWebSocketThread extends HandlerThread {
         public void run() {
           ddpClient.ping().continueWith(task -> {
             if (task.isFaulted()) {
-              RCLog.e(task.getError());
+              Exception error = task.getError();
+              RCLog.e(error);
               emitter.onSuccess(false);
-              ddpClient.close();
             } else {
               keepAliveTimer.update();
               emitter.onSuccess(true);
@@ -232,9 +240,26 @@ public class RocketChatWebSocketThread extends HandlerThread {
 
                 // handling WebSocket#onClose() callback.
                 task.getResult().client.getOnCloseCallback().onSuccess(_task -> {
-                  if (listenersRegistered) {
-                    terminate();
-                  }
+                  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)
+                    )
+                  );
                   return null;
                 });
 
@@ -265,6 +290,11 @@ public class RocketChatWebSocketThread extends HandlerThread {
         }));
   }
 
+  private void logErrorAndUnsubscribe(CompositeSubscription subscriptions, Throwable err) {
+    RCLog.e(err);
+    subscriptions.clear();
+  }
+
   @DebugLog
   private Single<Boolean> connect() {
     return connectDDPClient()
@@ -284,7 +314,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
     return new MethodCallHelper(realmHelper, ddpClientRef).getPermissions();
   }
 
-  //@DebugLog
+  @DebugLog
   private void registerListeners() {
     if (!Thread.currentThread().getName().equals("RC_thread_" + hostname)) {
       // execute in Looper.
@@ -293,7 +323,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
 
     if (listenersRegistered) {
-      return;
+      unregisterListeners();
     }
     listenersRegistered = true;
 
@@ -308,12 +338,30 @@ 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!!");
       }
     }
   }
 
+  @DebugLog
+  private void unregisterListenersAndClose() {
+   unregisterListeners();
+    if (ddpClient != null) {
+      ddpClient.close();
+      ddpClient = null;
+    }
+  }
+
   @DebugLog
   private void unregisterListeners() {
     Iterator<Registrable> iterator = listeners.iterator();
@@ -323,9 +371,5 @@ public class RocketChatWebSocketThread extends HandlerThread {
       iterator.remove();
     }
     listenersRegistered = false;
-    if (ddpClient != null) {
-      ddpClient.close();
-      ddpClient = null;
-    }
   }
 }

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

@@ -6,7 +6,7 @@ package chat.rocket.android.service;
 public class ServerConnectivity {
   public static final int STATE_CONNECTED = 1;
   public static final int STATE_DISCONNECTED = 2;
-  /*package*/ static final int STATE_CONNECTING = 3;
+  public static final int STATE_CONNECTING = 3;
   /*package*/ static final int STATE_DISCONNECTING = 4;
 
   public final String hostname;

app/src/main/java/chat/rocket/android/service/ddp/AbstractDDPDocEventSubscriber.java

@@ -185,7 +185,7 @@ public abstract class AbstractDDPDocEventSubscriber implements Registrable {
     if (rxSubscription != null) {
       rxSubscription.dispose();
     }
-    if (!TextUtils.isEmpty(subscriptionId)) {
+    if (!TextUtils.isEmpty(subscriptionId) && ddpClientRef.get() != null) {
       ddpClientRef.get().unsubscribe(subscriptionId).continueWith(new LogIfError());
     }
   }

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

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

app/src/main/java/chat/rocket/android/service/observer/MethodCallObserver.java

 package chat.rocket.android.service.observer;
 
 import android.content.Context;
-import io.realm.Realm;
-import io.realm.RealmResults;
+
 import org.json.JSONObject;
 
 import java.util.List;
+
 import chat.rocket.android.helper.CheckSum;
 import chat.rocket.android.helper.LogIfError;
-import chat.rocket.android_ddp.rx.RxWebSocketCallback;
-import chat.rocket.core.SyncState;
-import chat.rocket.persistence.realm.models.internal.MethodCall;
-import chat.rocket.persistence.realm.RealmHelper;
 import chat.rocket.android.service.DDPClientRef;
 import chat.rocket.android_ddp.DDPClientCallback;
+import chat.rocket.core.SyncState;
+import chat.rocket.persistence.realm.RealmHelper;
+import chat.rocket.persistence.realm.models.internal.MethodCall;
+import io.realm.Realm;
+import io.realm.RealmResults;
 
 /**
  * Observing MethodCall record, executing RPC if needed.
@@ -117,9 +118,6 @@ public class MethodCallObserver extends AbstractModelObserver<MethodCall> {
           } else if (exception instanceof DDPClientCallback.RPC.Timeout) {
             // temp "fix"- we need to rewrite the connection layer a bit
             errMessage = "{\"message\": \"Connection Timeout\"}";
-          } else if (exception instanceof RxWebSocketCallback.Failure) {
-            // temp "fix"- we need to rewrite the connection layer a bit
-            errMessage = "{\"message\": \"Connection Failure\"}";
           } else {
             errMessage = exception.getMessage();
           }

app/src/main/res/values/strings.xml

@@ -14,14 +14,20 @@
 
     <string name="start_of_conversation">Start of conversation</string>
     <string name="users_of_room_title">Members List</string>
-    <string name="fmt_room_user_count">Total: %,d users</string>
+    <plurals name="fmt_room_user_count">
+        <item quantity="one">Total: %,d user</item>
+        <item quantity="other">Total: %,d users</item>
+    </plurals>
     <string name="sending">Sending…</string>
     <string name="not_synced">Not synced</string>
     <string name="failed_to_sync">Failed to sync</string>
     <string name="resend">Resend</string>
     <string name="discard">Discard</string>
 
-    <string name="fmt_dialog_view_latest_message_title">New %d messages</string>
+    <plurals name="fmt_dialog_view_latest_message_title">
+        <item quantity="one">New %d message</item>
+        <item quantity="other">New %d messages</item>
+    </plurals>
     <string name="dialog_view_latest_message_action">View</string>
 
     <string name="file_uploading_title">Uploading…</string>

circle.yml

@@ -7,7 +7,7 @@
 machine:
   environment:
     ANDROID_HOME: /usr/local/android-sdk-linux
-    GRADLE_OPTS: '-Dorg.gradle.jvmargs="-Xmx2048m -XX:+HeapDumpOnOutOfMemoryError"'
+    GRADLE_OPTS: '-Dorg.gradle.jvmargs="-Xmx1536m -XX:MaxPermSize=512m -XX:+HeapDumpOnOutOfMemoryError"'
 
 dependencies:
   pre:

rocket-chat-android-widgets/src/main/java/chat/rocket/android/widget/RocketChatWidgets.java

 package chat.rocket.android.widget;
 
 import android.content.Context;
-import chat.rocket.android.widget.fresco.CustomImageFormatConfigurator;
+
 import com.facebook.common.logging.FLog;
 import com.facebook.drawee.backends.pipeline.DraweeConfig;
 import com.facebook.drawee.backends.pipeline.Fresco;
@@ -9,8 +9,11 @@ import com.facebook.imagepipeline.backends.okhttp3.OkHttpImagePipelineConfigFact
 import com.facebook.imagepipeline.core.ImagePipelineConfig;
 import com.facebook.imagepipeline.listener.RequestListener;
 import com.facebook.imagepipeline.listener.RequestLoggingListener;
+
 import java.util.HashSet;
 import java.util.Set;
+
+import chat.rocket.android.widget.fresco.CustomImageFormatConfigurator;
 import okhttp3.OkHttpClient;
 
 public class RocketChatWidgets {