A change of network type will now show a snackbar when a failure occur which...

A change of network type will now show a snackbar when a failure occur which will trigger a websocket reconnection flow

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

@@ -61,11 +61,7 @@ public class DDPClient {
     return impl.getOnCloseCallback();
   }
 
-  public Task<RxWebSocketCallback.Failure> getOnFailureCallback() {
-    return impl.getOnFailureCallback();
-  }
-
-  public void close(int code, String reason) {
-    impl.close(code, reason);
+  public void close() {
+    impl.close(1000, "closed by DDPClient#close()");
   }
 }

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

@@ -59,7 +59,7 @@ public class DDPClientImpl {
   public void connect(final TaskCompletionSource<DDPClientCallback.Connect> task, final String url,
                       String session) {
     try {
-      flowable = websocket.connect(url).autoConnect();
+      flowable = websocket.connect(url).autoConnect(2);
       CompositeDisposable disposables = new CompositeDisposable();
 
       disposables.add(
@@ -125,7 +125,6 @@ public class DDPClientImpl {
           flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
               .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
               .map(DDPClientImpl::toJson)
-              .timeout(4, TimeUnit.SECONDS)
               .subscribe(
                   response -> {
                     String msg = extractMsg(response);
@@ -370,29 +369,14 @@ public class DDPClientImpl {
     });
   }
 
-  public Task<RxWebSocketCallback.Failure> getOnFailureCallback() {
-    TaskCompletionSource<RxWebSocketCallback.Failure> task = new TaskCompletionSource<>();
-
-    flowable.filter(callback -> callback instanceof RxWebSocketCallback.Failure)
-            .cast(RxWebSocketCallback.Failure.class)
-            .subscribe(
-                    task::setResult,
-                    err -> setTaskError(task, err)
-            );
-
-    return task.getTask().onSuccessTask(_task -> {
-      unsubscribeBaseListeners();
-      return _task;
-    });
-  }
-
   private boolean sendMessage(String msg, @Nullable JSONBuilder json) {
     try {
       JSONObject origJson = new JSONObject().put("msg", msg);
       String msg2 = (json == null ? origJson : json.create(origJson)).toString();
-      return websocket.sendText(msg2);
+      websocket.sendText(msg2);
     } catch (Exception e) {
       RCLog.e(e);
+      return false;
     }
     return true; // ignore exception here.
   }

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

 package chat.rocket.android_ddp.rx;
 
 import java.io.IOException;
-import java.net.UnknownHostException;
 import java.util.concurrent.TimeUnit;
 
 import chat.rocket.android.log.RCLog;
@@ -18,8 +17,13 @@ import okhttp3.WebSocket;
 import okhttp3.WebSocketListener;
 
 public class RxWebSocket {
+  public static final int REASON_CLOSED_BY_USER = 101;
+  public static final int REASON_NETWORK_ERROR = 102;
+  public static final int REASON_SERVER_ERROR = 103;
+  public static final int REASON_UNKNOWN = 104;
   private OkHttpClient httpClient;
   private WebSocket webSocket;
+  private boolean hadErrorsBefore;
 
   public RxWebSocket(OkHttpClient client) {
     httpClient = client;
@@ -33,6 +37,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));
               }
@@ -40,10 +45,10 @@ public class RxWebSocket {
               @Override
               public void onFailure(WebSocket webSocket, Throwable err, Response response) {
                 try {
-                  if (err instanceof UnknownHostException) {
-                    emitter.onError(err);
-                  } else {
-                    emitter.onNext(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");
@@ -71,7 +76,7 @@ public class RxWebSocket {
               }
             }),
         BackpressureStrategy.BUFFER
-    ).delay(2000, TimeUnit.MILLISECONDS).publish();
+    ).delay(4, TimeUnit.SECONDS).publish();
   }
 
   public boolean sendText(String message) throws IOException {

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

@@ -20,7 +20,6 @@ import hu.akarnokd.rxjava.interop.RxJavaInterop;
 import io.reactivex.Flowable;
 import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.schedulers.Schedulers;
 
 public class MainPresenter extends BasePresenter<MainContract.View>
     implements MainContract.Presenter {
@@ -65,6 +64,7 @@ public class MainPresenter extends BasePresenter<MainContract.View>
 
     openRoom();
 
+    subscribeToNetworkChanges();
     subscribeToUnreadCount();
     subscribeToSession();
     setUserOnline();
@@ -98,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,19 +159,18 @@ public class MainPresenter extends BasePresenter<MainContract.View>
         );
 
     addSubscription(subscription);
+  }
 
-    // Update to RxJava 2 (issue: https://github.com/RocketChat/Rocket.Chat.Android/issues/355)
-    addSubscription(
-            RxJavaInterop.toV2Observable(connectivityManagerApi.getServerConnectivityAsObservable())
-                    .subscribeOn(Schedulers.io())
+  private void subscribeToNetworkChanges() {
+    Disposable disposable = RxJavaInterop.toV2Flowable(connectivityManagerApi.getServerConnectivityAsObservable())
+            .filter(connectivity -> connectivity.state == ServerConnectivity.STATE_DISCONNECTED)
             .observeOn(AndroidSchedulers.mainThread())
-                    .subscribe(serverConnectivity -> {
-                      if (serverConnectivity.state == ServerConnectivity.STATE_CONNECTING) {
-                        view.showConnecting();
-                      }
-                    },
-                    Logger::report)
+            .subscribe(
+                    a -> view.showConnectionError(),
+                    Logger::report
             );
+
+    addSubscription(disposable);
   }
 
   private void setUserOnline() {

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.
@@ -45,8 +47,8 @@ public class DDPClientWrapper {
   /**
    * close connection.
    */
-  public void close(int code, String reason) {
-    ddpClient.close(code, reason);
+  public void close() {
+    ddpClient.close();
   }
 
   /**

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;
 
@@ -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/service/RocketChatWebSocketThread.java

@@ -32,14 +32,11 @@ 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.DDPClientImpl;
 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 hu.akarnokd.rxjava.interop.RxJavaInterop;
 import hugo.weaving.DebugLog;
-import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.disposables.CompositeDisposable;
 import rx.Single;
 
@@ -68,7 +65,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
   private final RealmHelper realmHelper;
   private final ConnectivityManagerInternal connectivityManager;
   private final ArrayList<Registrable> listeners = new ArrayList<>();
-  private DDPClientWrapper ddpClient;
+  private static DDPClientWrapper ddpClient;
   private boolean listenersRegistered;
   private RocketChatCache rocketChatCache;
   private DDPClientRef ddpClientRef = new DDPClientRef() {
@@ -157,7 +154,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();
@@ -206,9 +203,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(DDPClientImpl.CLOSED_NOT_ALIVE, "Ping timeout");
             } else {
               keepAliveTimer.update();
               emitter.onSuccess(true);
@@ -248,24 +245,10 @@ public class RocketChatWebSocketThread extends HandlerThread {
 
                 // handling WebSocket#onClose() callback.
                 task.getResult().client.getOnCloseCallback().onSuccess(_task -> {
-                  if (listenersRegistered) {
-                    terminate();
-                  }
-                  return null;
-                });
-
-                task.getResult().client.getOnFailureCallback().onSuccess(_task -> {
+                  ddpClient.close();
                   ddpClient = null;
-                  CompositeDisposable disposables = new CompositeDisposable();
-                  connectivityManager.notifyConnecting(hostname);
-                  disposables.add(
-                          RxJavaInterop.toV2Single(connect().retry())
-                                  .observeOn(AndroidSchedulers.from(getLooper()))
-                                  .subscribe(
-                                          rocketChatWebSocketThread -> forceRegisteringListeners(),
-                                          err -> logErrorAndDispose(err, disposables)
-                                  )
-                  );
+                  connectivityManager.notifyConnectionLost(hostname,
+                          ConnectivityManagerInternal.REASON_NETWORK_ERROR);
                   return null;
                 });
 
@@ -331,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.
@@ -340,7 +323,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
 
     if (listenersRegistered) {
-      return;
+      unregisterListeners();
     }
     listenersRegistered = true;
 
@@ -355,12 +338,12 @@ public class RocketChatWebSocketThread extends HandlerThread {
           registrable.register();
           listeners.add(registrable);
         }
-
+        // Register for room stream messages
         String roomId = rocketChatCache.getSelectedRoomId();
         if (roomId != null) {
           StreamRoomMessage streamRoomMessage = new StreamRoomMessage(
-                  appContext, hostname, realmHelper, ddpClientRef, roomId);
-
+                  appContext, hostname, realmHelper, ddpClientRef, roomId
+          );
           streamRoomMessage.register();
           listeners.add(streamRoomMessage);
         }
@@ -370,6 +353,15 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
   }
 
+  @DebugLog
+  private void unregisterListenersAndClose() {
+   unregisterListeners();
+    if (ddpClient != null) {
+      ddpClient.close();
+      ddpClient = null;
+    }
+  }
+
   @DebugLog
   private void unregisterListeners() {
     Iterator<Registrable> iterator = listeners.iterator();
@@ -379,9 +371,5 @@ public class RocketChatWebSocketThread extends HandlerThread {
       iterator.remove();
     }
     listenersRegistered = false;
-    if (ddpClient != null) {
-      ddpClient.close(DDPClientImpl.CLOSED_NORMALLY, "Closed by client");
-      ddpClient = null;
-    }
   }
 }