Merge pull request #586 from RocketChat/fix/indefinite-status-ticker

[FIX] Infinite status ticker

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

@@ -50,30 +50,30 @@ public class DDPClient {
         impl = new DDPClientImpl(this, client);
     }
 
-    public Task<DDPClientCallback.Connect> connect(String url, String session) {
+    private Task<DDPClientCallback.Connect> connect(String url, String session) {
         hostname.set(url);
         TaskCompletionSource<DDPClientCallback.Connect> task = new TaskCompletionSource<>();
         impl.connect(task, url, session);
         return task.getTask();
     }
 
-    public Task<DDPClientCallback.Ping> ping(@Nullable String id) {
+    private Task<DDPClientCallback.Ping> ping(@Nullable String id) {
         TaskCompletionSource<DDPClientCallback.Ping> task = new TaskCompletionSource<>();
         impl.ping(task, id);
         return task.getTask();
     }
 
-    public Maybe<DDPClientCallback.Base> doPing(@Nullable String id) {
+    private Maybe<DDPClientCallback.Base> doPing(@Nullable String id) {
         return impl.ping(id);
     }
 
-    public Task<DDPSubscription.Ready> sub(String id, String name, JSONArray params) {
+    private Task<DDPSubscription.Ready> sub(String id, String name, JSONArray params) {
         TaskCompletionSource<DDPSubscription.Ready> task = new TaskCompletionSource<>();
         impl.sub(task, name, params, id);
         return task.getTask();
     }
 
-    public Task<DDPSubscription.NoSub> unsub(String id) {
+    private Task<DDPSubscription.NoSub> unsub(String id) {
         TaskCompletionSource<DDPSubscription.NoSub> task = new TaskCompletionSource<>();
         impl.unsub(task, id);
         return task.getTask();

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

@@ -52,7 +52,7 @@ public class DDPClientImpl {
     }
   }
 
-  public void connect(final TaskCompletionSource<DDPClientCallback.Connect> task, final String url,
+  /* package */ void connect(final TaskCompletionSource<DDPClientCallback.Connect> task, final String url,
                       String session) {
     try {
       flowable = websocket.connect(url).autoConnect(2);

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

@@ -47,9 +47,9 @@ import hugo.weaving.DebugLog;
  */
 public class MainActivity extends AbstractAuthedActivity implements MainContract.View {
     private RoomToolbar toolbar;
-    private StatusTicker statusTicker;
     private SlidingPaneLayout pane;
     private MainContract.Presenter presenter;
+    private volatile Snackbar statusTicker;
 
     @Override
     public int getLayoutContainerForFragment() {
@@ -61,7 +61,6 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
         super.onCreate(savedInstanceState);
         setContentView(R.layout.activity_main);
         toolbar = findViewById(R.id.activity_main_toolbar);
-        statusTicker = new StatusTicker();
         pane = findViewById(R.id.sliding_pane);
         setupToolbar();
     }
@@ -95,6 +94,8 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
         if (presenter != null) {
             presenter.release();
         }
+        // Dismiss any status ticker
+        if (statusTicker != null) statusTicker.dismiss();
 
         super.onPause();
     }
@@ -245,28 +246,36 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
     @Override
     public void showLoginScreen() {
         LaunchUtil.showLoginActivity(this, hostname);
-        statusTicker.updateStatus(StatusTicker.STATUS_DISMISS, null);
+        showConnectionOk();
     }
 
     @Override
-    public void showConnectionError() {
-        statusTicker.updateStatus(StatusTicker.STATUS_CONNECTION_ERROR,
-                Snackbar.make(findViewById(getLayoutContainerForFragment()),
+    public synchronized void showConnectionError() {
+        dismissStatusTickerIfShowing();
+        statusTicker = Snackbar.make(findViewById(getLayoutContainerForFragment()),
                 R.string.fragment_retry_login_error_title, Snackbar.LENGTH_INDEFINITE)
                 .setAction(R.string.fragment_retry_login_retry_title, view ->
-                                ConnectivityManager.getInstance(getApplicationContext()).keepAliveServer()));
+                        ConnectivityManager.getInstance(getApplicationContext()).keepAliveServer());
+        statusTicker.show();
     }
 
     @Override
-    public void showConnecting() {
-        statusTicker.updateStatus(StatusTicker.STATUS_TOKEN_LOGIN,
-                Snackbar.make(findViewById(getLayoutContainerForFragment()),
-                        R.string.server_config_activity_authenticating, Snackbar.LENGTH_INDEFINITE));
+    public synchronized void showConnecting() {
+        dismissStatusTickerIfShowing();
+        statusTicker = Snackbar.make(findViewById(getLayoutContainerForFragment()),
+                R.string.server_config_activity_authenticating, Snackbar.LENGTH_INDEFINITE);
+        statusTicker.show();
     }
 
     @Override
-    public void showConnectionOk() {
-        statusTicker.updateStatus(StatusTicker.STATUS_DISMISS, null);
+    public synchronized void showConnectionOk() {
+        dismissStatusTickerIfShowing();
+    }
+
+    private void dismissStatusTickerIfShowing() {
+        if (statusTicker != null) {
+            statusTicker.dismiss();
+        }
     }
 
     @Override
@@ -349,34 +358,4 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
     public void beforeLogoutCleanUp() {
         presenter.beforeLogoutCleanUp();
     }
-
-    //TODO: consider this class to define in layouthelper for more complicated operation.
-    private static class StatusTicker {
-        static final int STATUS_DISMISS = 0;
-        static final int STATUS_CONNECTION_ERROR = 1;
-        static final int STATUS_TOKEN_LOGIN = 2;
-
-        private int status;
-        private Snackbar snackbar;
-
-        StatusTicker() {
-            status = STATUS_DISMISS;
-        }
-
-        void updateStatus(int status, Snackbar snackbar) {
-            if (status == this.status) {
-                return;
-            }
-            this.status = status;
-            if (this.snackbar != null) {
-                this.snackbar.dismiss();
-            }
-            if (status != STATUS_DISMISS) {
-                this.snackbar = snackbar;
-                if (this.snackbar != null) {
-                    this.snackbar.show();
-                }
-            }
-        }
-    }
 }

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

@@ -220,6 +220,7 @@ public class MainPresenter extends BasePresenter<MainContract.View>
 
     private void subscribeToNetworkChanges() {
         Disposable disposable = connectivityManagerApi.getServerConnectivityAsObservable()
+                .distinctUntilChanged()
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(
                         connectivity -> {

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

@@ -6,7 +6,7 @@ import android.support.annotation.Nullable;
 import java.util.List;
 
 import chat.rocket.core.models.ServerInfo;
-import io.reactivex.Observable;
+import io.reactivex.Flowable;
 import io.reactivex.Single;
 
 /**
@@ -23,7 +23,7 @@ public interface ConnectivityManagerApi {
 
     List<ServerInfo> getServerList();
 
-  Observable<ServerConnectivity> getServerConnectivityAsObservable();
+    Flowable<ServerConnectivity> getServerConnectivityAsObservable();
 
     int getConnectivityState(@NonNull String hostname);
 

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

@@ -8,10 +8,6 @@ import chat.rocket.core.models.ServerInfo;
  * interfaces used for RocketChatService and RocketChatwebSocketThread.
  */
 /*package*/ interface ConnectivityManagerInternal {
-  int REASON_CLOSED_BY_USER = 101;
-  int REASON_NETWORK_ERROR = 102;
-  int REASON_SERVER_ERROR = 103;
-  int REASON_UNKNOWN = 104;
 
   void resetConnectivityStateList();
 

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

@@ -21,10 +21,11 @@ import chat.rocket.android_ddp.DDPClient;
 import chat.rocket.core.models.ServerInfo;
 import chat.rocket.persistence.realm.models.RealmBasedServerInfo;
 import hugo.weaving.DebugLog;
-import io.reactivex.Observable;
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
 import io.reactivex.Single;
 import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subjects.PublishSubject;
+import io.reactivex.subjects.BehaviorSubject;
 
 /**
  * Connectivity management implementation.
@@ -32,7 +33,7 @@ import io.reactivex.subjects.PublishSubject;
 /*package*/ class RealmBasedConnectivityManager
         implements ConnectivityManagerApi, ConnectivityManagerInternal {
     private volatile ConcurrentHashMap<String, Integer> serverConnectivityList = new ConcurrentHashMap<>();
-  private final PublishSubject<ServerConnectivity> connectivitySubject = PublishSubject.create();
+    private volatile BehaviorSubject<ServerConnectivity> connectivitySubject = BehaviorSubject.createDefault(ServerConnectivity.CONNECTED);
     private Context appContext;
     private final ServiceConnection serviceConnection = new ServiceConnection() {
         @Override
@@ -79,7 +80,10 @@ import io.reactivex.subjects.PublishSubject;
         }
         connectToServerIfNeeded(hostname, true/* force connect */)
                 .subscribeOn(Schedulers.io())
-          .subscribe(_val -> {
+                .subscribe(connected -> {
+                    if (!connected) {
+                        notifyConnectionLost(hostname, DDPClient.REASON_NETWORK_ERROR);
+                    }
                 }, error -> {
                     RCLog.e(error);
                     notifyConnectionLost(hostname, DDPClient.REASON_NETWORK_ERROR);
@@ -94,7 +98,7 @@ import io.reactivex.subjects.PublishSubject;
             serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTED);
         }
         connectToServerIfNeeded(hostname, false)
-        .subscribe(_val -> {
+                .subscribe(connected -> {
                 }, RCLog::e);
     }
 
@@ -135,7 +139,9 @@ import io.reactivex.subjects.PublishSubject;
     @DebugLog
     @Override
     public void notifyConnectionEstablished(String hostname, String session) {
+        if (session != null) {
             RealmBasedServerInfo.updateSession(hostname, session);
+        }
         serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTED);
         connectivitySubject.onNext(
                 new ServerConnectivity(hostname, ServerConnectivity.STATE_CONNECTED));
@@ -158,8 +164,8 @@ import io.reactivex.subjects.PublishSubject;
     }
 
     @Override
-  public Observable<ServerConnectivity> getServerConnectivityAsObservable() {
-    return Observable.concat(Observable.fromIterable(getCurrentConnectivityList()), connectivitySubject);
+    public Flowable<ServerConnectivity> getServerConnectivityAsObservable() {
+        return connectivitySubject.toFlowable(BackpressureStrategy.LATEST);
     }
 
     @Override
@@ -185,10 +191,12 @@ import io.reactivex.subjects.PublishSubject;
             }
 
             if (connectivity == ServerConnectivity.STATE_DISCONNECTED) {
-        notifyConnecting(hostname);
+//              notifyConnecting(hostname);
             }
 
-      return connectToServer(hostname);
+            return connectToServer(hostname)
+                    .retry(exception -> exception instanceof ThreadLooperNotPreparedException)
+                    .onErrorResumeNext(Single.just(false));
         });
     }
 
@@ -257,7 +265,7 @@ import io.reactivex.subjects.PublishSubject;
             if (serviceInterface != null) {
                 return serviceInterface.ensureConnectionToServer(hostname);
             } else {
-        return Single.error(new IllegalStateException("not prepared"));
+                return Single.error(new ThreadLooperNotPreparedException("not prepared"));
             }
         });
     }
@@ -278,4 +286,10 @@ import io.reactivex.subjects.PublishSubject;
             }
         });
     }
+
+    private static class ThreadLooperNotPreparedException extends IllegalStateException {
+        ThreadLooperNotPreparedException(String message) {
+            super(message);
+        }
+    }
 }
\ No newline at end of file

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

@@ -38,7 +38,7 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
     /**
      * ensure RocketChatService alive.
      */
-  /*package*/ static void keepAlive(Context context) {
+    /*package*/static void keepAlive(Context context) {
         context.startService(new Intent(context, RocketChatService.class));
     }
 
@@ -69,11 +69,7 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
     @Override
     public Single<Boolean> ensureConnectionToServer(String hostname) { //called via binder.
         return getOrCreateWebSocketThread(hostname)
-        .doOnError(err -> {
-          err.printStackTrace();
-          currentWebSocketThread = null;
-        })
-        .flatMap(webSocketThreads -> webSocketThreads.keepAlive());
+                .flatMap(RocketChatWebSocketThread::keepAlive);
     }
 
     @Override
@@ -109,12 +105,9 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
                 return Single.just(currentWebSocketThread);
             }
 
-      connectivityManager.notifyConnecting(hostname);
-
             if (currentWebSocketThread != null) {
                 return currentWebSocketThread.terminate()
                         .doAfterTerminate(() -> currentWebSocketThread = null)
-              .doOnError(RCLog::e)
                         .flatMap(terminated ->
                                 RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
                                         .doOnSuccess(thread -> {

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

@@ -76,26 +76,6 @@ public class RocketChatWebSocketThread extends HandlerThread {
     private final CompositeDisposable reconnectDisposable = new CompositeDisposable();
     private boolean listenersRegistered;
 
-    private static class KeepAliveTimer {
-        private long lastTime;
-        private final long thresholdMs;
-
-        public KeepAliveTimer(long thresholdMs) {
-            this.thresholdMs = thresholdMs;
-            lastTime = System.currentTimeMillis();
-        }
-
-        public boolean shouldCheckPrecisely() {
-            return lastTime + thresholdMs < System.currentTimeMillis();
-        }
-
-        public void update() {
-            lastTime = System.currentTimeMillis();
-        }
-    }
-
-    private final KeepAliveTimer keepAliveTimer = new KeepAliveTimer(20000);
-
     private RocketChatWebSocketThread(Context appContext, String hostname) {
         super("RC_thread_" + hostname);
         this.appContext = appContext;
@@ -108,7 +88,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
      * build new Thread.
      */
     @DebugLog
-    public static Single<RocketChatWebSocketThread> getStarted(Context appContext, String hostname) {
+    /* package */ static Single<RocketChatWebSocketThread> getStarted(Context appContext, String hostname) {
         return Single.<RocketChatWebSocketThread>create(objectSingleEmitter -> {
             new RocketChatWebSocketThread(appContext, hostname) {
                 @Override
@@ -148,7 +128,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
      * terminate WebSocket thread.
      */
     @DebugLog
-    public Single<Boolean> terminate() {
+    /* package */ Single<Boolean> terminate() {
         if (isAlive()) {
             return Single.create(emitter -> {
                 new Handler(getLooper()).post(() -> {
@@ -181,7 +161,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
      * synchronize the state of the thread with ServerConfig.
      */
     @DebugLog
-    public Single<Boolean> keepAlive() {
+    /* package */ Single<Boolean> keepAlive() {
         return checkIfConnectionAlive()
                 .flatMap(alive -> alive ? Single.just(true) : connectWithExponentialBackoff());
     }
@@ -192,11 +172,6 @@ public class RocketChatWebSocketThread extends HandlerThread {
             return Single.just(false);
         }
 
-        if (!keepAliveTimer.shouldCheckPrecisely()) {
-            return Single.just(true);
-        }
-        keepAliveTimer.update();
-
         return Single.create(emitter -> {
             new Thread() {
                 @Override
@@ -207,9 +182,8 @@ public class RocketChatWebSocketThread extends HandlerThread {
                             RCLog.e(error);
                             connectivityManager.notifyConnectionLost(
                                     hostname, DDPClient.REASON_CLOSED_BY_USER);
-                            emitter.onError(error);
+                            emitter.onSuccess(false);
                         } else {
-                            keepAliveTimer.update();
                             emitter.onSuccess(true);
                         }
                         return null;
@@ -245,11 +219,11 @@ public class RocketChatWebSocketThread extends HandlerThread {
                 return;
             }
             RCLog.d("DDPClient#connect");
+            connectivityManager.notifyConnecting(hostname);
             DDPClient.get().connect(hostname, info.getSession(), info.isSecure())
                     .onSuccessTask(task -> {
                         final String newSession = task.getResult().session;
                         connectivityManager.notifyConnectionEstablished(hostname, newSession);
-
                         // handling WebSocket#onClose() callback.
                         task.getResult().client.getOnCloseCallback().onSuccess(_task -> {
                             RxWebSocketCallback.Close result = _task.getResult();
@@ -292,18 +266,18 @@ public class RocketChatWebSocketThread extends HandlerThread {
             return;
         }
         forceInvalidateTokens();
-        connectivityManager.notifyConnecting(hostname);
         reconnectDisposable.add(
                 connectWithExponentialBackoff()
                         .subscribe(connected -> {
                                     if (!connected) {
-                                        connectivityManager.notifyConnecting(hostname);
+                                        connectivityManager.notifyConnectionLost(hostname,
+                                                DDPClient.REASON_NETWORK_ERROR);
                                     }
                                     reconnectDisposable.clear();
                                 }, error -> {
-                                    logErrorAndUnsubscribe(reconnectDisposable, error);
                                     connectivityManager.notifyConnectionLost(hostname,
                                             DDPClient.REASON_NETWORK_ERROR);
+                                    logErrorAndUnsubscribe(reconnectDisposable, error);
                                 }
                         )
         );
@@ -315,7 +289,9 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
 
     private Single<Boolean> connectWithExponentialBackoff() {
-        return connect().retryWhen(RxHelper.exponentialBackoff(3, 500, TimeUnit.MILLISECONDS));
+        return connect()
+                .retryWhen(RxHelper.exponentialBackoff(1, 250, TimeUnit.MILLISECONDS))
+                .onErrorResumeNext(Single.just(false));
     }
 
     @DebugLog

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

@@ -4,10 +4,13 @@ package chat.rocket.android.service;
  * pair with server's hostname and its connectivity state.
  */
 public class ServerConnectivity {
+
   public static final int STATE_CONNECTED = 1;
+
   public static final int STATE_DISCONNECTED = 2;
   public static final int STATE_CONNECTING = 3;
   /*package*/ static final int STATE_DISCONNECTING = 4;
+  public static final ServerConnectivity CONNECTED = new ServerConnectivity(null, STATE_CONNECTED);
 
   public final String hostname;
   public final int state;
@@ -25,6 +28,21 @@ public class ServerConnectivity {
     this.code = code;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+
+    ServerConnectivity that = (ServerConnectivity) o;
+
+    return state == that.state;
+  }
+
+  @Override
+  public int hashCode() {
+    return state;
+  }
+
   /**
    * This exception should be thrown when connection is lost during waiting for CONNECTED.
    */