Get rid of StatusTicker internal class on MainActivity and notify more...

Get rid of StatusTicker internal class on MainActivity and notify more appropriately when there is a connection change

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 -> {
@@ -229,6 +230,8 @@ public class MainPresenter extends BasePresenter<MainContract.View>
                       } else if (connectivity.state == ServerConnectivity.STATE_DISCONNECTED) {
                           if (connectivity.code == DDPClient.REASON_NETWORK_ERROR) {
                               view.showConnectionError();
+                          } else {
+                              view.showConnectionOk();
                           }
                       } else {
                           view.showConnecting();

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/RealmBasedConnectivityManager.java

@@ -21,9 +21,12 @@ 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.BackpressureStrategy;
+import io.reactivex.Flowable;
 import io.reactivex.Observable;
 import io.reactivex.Single;
 import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subjects.BehaviorSubject;
 import io.reactivex.subjects.PublishSubject;
 
 /**
@@ -32,7 +35,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
@@ -135,7 +138,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 +163,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,7 +190,7 @@ import io.reactivex.subjects.PublishSubject;
       }
 
       if (connectivity == ServerConnectivity.STATE_DISCONNECTED) {
-        notifyConnecting(hostname);
+//        notifyConnecting(hostname);
       }
 
       return connectToServer(hostname);

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

@@ -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

@@ -245,11 +245,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 +292,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 +315,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.
    */