Handling timeout errors from service keep alive as a Failure callback event instead of a close one

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

@@ -65,7 +65,7 @@ public class DDPClient {
     return impl.getOnFailureCallback();
   }
 
-  public void close() {
-    impl.close(1000, "closed by DDPClient#close()");
+  public void close(int code, String reason) {
+    impl.close(code, reason);
   }
 }

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

 package chat.rocket.android_ddp;
 
+import android.os.SystemClock;
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 import android.text.TextUtils;
@@ -8,6 +9,7 @@ import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
 
+import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -19,8 +21,11 @@ import chat.rocket.android_ddp.rx.RxWebSocketCallback;
 import io.reactivex.Flowable;
 import io.reactivex.disposables.CompositeDisposable;
 import okhttp3.OkHttpClient;
+import okhttp3.Response;
 
 public class DDPClientImpl {
+  public static final int CLOSED_NORMALLY = 1000;
+  public static final int CLOSED_NOT_ALIVE = 1001;
   private final DDPClient client;
   private RxWebSocket websocket;
   private Flowable<RxWebSocketCallback.Base> flowable;
@@ -135,6 +140,7 @@ public class DDPClientImpl {
                           disposables.clear();
                         }
                       }
+                      disposables.clear();
                     }
                   },
                   err -> task.setError(new DDPClientCallback.Ping.Timeout(client))
@@ -286,6 +292,7 @@ public class DDPClientImpl {
                 response -> {
                   String msg = extractMsg(response);
                   if ("ping".equals(msg)) {
+                    SystemClock.sleep(8000);
                     if (response.isNull("id")) {
                       sendMessage("pong", null);
                     } else {

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

 package chat.rocket.android_ddp.rx;
 
-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 java.net.UnknownHostException;
 import java.util.concurrent.TimeUnit;
 
 import chat.rocket.android.log.RCLog;
+import chat.rocket.android_ddp.DDPClientImpl;
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableOnSubscribe;
+import io.reactivex.exceptions.OnErrorNotImplementedException;
+import io.reactivex.flowables.ConnectableFlowable;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
@@ -57,8 +57,17 @@ public class RxWebSocket {
 
               @Override
               public void onClosed(WebSocket webSocket, int code, String reason) {
-                emitter.onNext(new RxWebSocketCallback.Close(webSocket, code, reason));
-                emitter.onComplete();
+                switch (code) {
+                  case DDPClientImpl.CLOSED_NORMALLY:
+                    emitter.onNext(new RxWebSocketCallback.Close(webSocket, code, reason));
+                    emitter.onComplete();
+                    break;
+                  case DDPClientImpl.CLOSED_NOT_ALIVE:
+                    emitter.onNext(new RxWebSocketCallback.Failure(webSocket, new Exception(reason), null));
+                    break;
+                  default:
+                    RCLog.e("Websocket closed abnormally");
+                }
               }
             }),
         BackpressureStrategy.BUFFER

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

@@ -45,8 +45,8 @@ public class DDPClientWrapper {
   /**
    * close connection.
    */
-  public void close() {
-    ddpClient.close();
+  public void close(int code, String reason) {
+    ddpClient.close(code, reason);
   }
 
   /**

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

@@ -32,6 +32,7 @@ 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;
@@ -207,7 +208,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
             if (task.isFaulted()) {
               RCLog.e(task.getError());
               emitter.onSuccess(false);
-              ddpClient.close();
+              ddpClient.close(DDPClientImpl.CLOSED_NOT_ALIVE, "Ping timeout");
             } else {
               keepAliveTimer.update();
               emitter.onSuccess(true);
@@ -379,7 +380,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
     listenersRegistered = false;
     if (ddpClient != null) {
-      ddpClient.close();
+      ddpClient.close(DDPClientImpl.CLOSED_NORMALLY, "Closed by client");
       ddpClient = null;
     }
   }