add error handling for "already closed"

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

@@ -15,7 +15,8 @@ public class DDPClientCallback {
   public static abstract class BaseException extends Exception {
     public DDPClient client;
 
-    public BaseException(DDPClient client) {
+    public BaseException(Class<? extends BaseException> clazz, DDPClient client) {
+      super(clazz.getName());
       this.client = client;
     }
   }
@@ -32,14 +33,14 @@ public class DDPClientCallback {
       public String version;
 
       public Failed(DDPClient client, String version) {
-        super(client);
+        super(Failed.class, client);
         this.version = version;
       }
     }
 
     public static class Timeout extends BaseException {
       public Timeout(DDPClient client) {
-        super(client);
+        super(Timeout.class, client);
       }
     }
   }
@@ -54,7 +55,7 @@ public class DDPClientCallback {
 
     public static class Timeout extends BaseException {
       public Timeout(DDPClient client) {
-        super(client);
+        super(Timeout.class, client);
       }
     }
   }
@@ -74,7 +75,7 @@ public class DDPClientCallback {
       public JSONObject error;
 
       public Error(DDPClient client, String id, JSONObject error) {
-        super(client);
+        super(Error.class, client);
         this.id = id;
         this.error = error;
       }
@@ -82,8 +83,14 @@ public class DDPClientCallback {
 
     public static class Timeout extends BaseException {
       public Timeout(DDPClient client) {
-        super(client);
+        super(Timeout.class, client);
+      }
     }
   }
+
+  public static class Closed extends BaseException {
+    public Closed(DDPClient client) {
+      super(Closed.class, client);
+    }
   }
 }

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

@@ -59,7 +59,8 @@ public class DDPClientImpl {
           .subscribe(callback -> {
             sendMessage("connect",
                 json -> (TextUtils.isEmpty(session) ? json : json.put("session", session)).put(
-                    "version", "pre2").put("support", new JSONArray().put("pre2").put("pre1")));
+                    "version", "pre2").put("support", new JSONArray().put("pre2").put("pre1")),
+                task);
           }, err -> {
           }));
 
@@ -71,20 +72,20 @@ public class DDPClientImpl {
               .subscribe(response -> {
                 String msg = extractMsg(response);
                 if ("connected".equals(msg) && !response.isNull("session")) {
-                  task.setResult(
+                  task.trySetResult(
                       new DDPClientCallback.Connect(client, response.optString("session")));
                   subscriptions.unsubscribe();
                 } else if ("error".equals(msg) && "Already connected".equals(
                     response.optString("reason"))) {
-                  task.setResult(new DDPClientCallback.Connect(client, null));
+                  task.trySetResult(new DDPClientCallback.Connect(client, null));
                   subscriptions.unsubscribe();
                 } else if ("failed".equals(msg)) {
-                  task.setError(
+                  task.trySetError(
                       new DDPClientCallback.Connect.Failed(client, response.optString("version")));
                   subscriptions.unsubscribe();
                 }
               }, err -> {
-                task.setError(new DDPClientCallback.Connect.Timeout(client));
+                task.trySetError(new DDPClientCallback.Connect.Timeout(client));
               }));
 
       addErrorCallback(subscriptions, task);
@@ -101,6 +102,12 @@ public class DDPClientImpl {
 
   public void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
                    @Nullable final String id) {
+
+    final boolean requested = (TextUtils.isEmpty(id)) ?
+        sendMessage("ping", null) :
+        sendMessage("ping", json -> json.put("id", id));
+
+    if (requested) {
       CompositeSubscription subscriptions = new CompositeSubscription();
 
       subscriptions.add(
@@ -127,16 +134,15 @@ public class DDPClientImpl {
               }));
 
       addErrorCallback(subscriptions, task);
-
-    if (TextUtils.isEmpty(id)) {
-      sendMessage("ping", null);
-    } else {
-      sendMessage("ping", json -> json.put("id", id));
     }
   }
 
   public void sub(final TaskCompletionSource<DDPSubscription.Ready> task, String name,
                   JSONArray params, String id) {
+    final boolean requested =
+        sendMessage("sub", json -> json.put("id", id).put("name", name).put("params", params));
+
+    if (requested) {
       CompositeSubscription subscriptions = new CompositeSubscription();
 
       subscriptions.add(
@@ -168,12 +174,15 @@ public class DDPClientImpl {
               }));
 
       addErrorCallback(subscriptions, task);
-
-    sendMessage("sub", json -> json.put("id", id).put("name", name).put("params", params));
+    }
   }
 
   public void unsub(final TaskCompletionSource<DDPSubscription.NoSub> task,
                     @Nullable final String id) {
+
+    final boolean requested = sendMessage("unsub", json -> json.put("id", id));
+
+    if (requested) {
       CompositeSubscription subscriptions = new CompositeSubscription();
 
       subscriptions.add(
@@ -193,12 +202,16 @@ public class DDPClientImpl {
               }));
 
       addErrorCallback(subscriptions, task);
-
-    sendMessage("unsub", json -> json.put("id", id));
+    }
   }
 
   public void rpc(final TaskCompletionSource<DDPClientCallback.RPC> task, String method,
                   JSONArray params, String id, long timeoutMs) {
+    final boolean requested =
+        sendMessage("method",
+            json -> json.put("method", method).put("params", params).put("id", id));
+
+    if (requested) {
       CompositeSubscription subscriptions = new CompositeSubscription();
 
       subscriptions.add(
@@ -228,8 +241,7 @@ public class DDPClientImpl {
               }));
 
       addErrorCallback(subscriptions, task);
-
-    sendMessage("method", json -> json.put("method", method).put("params", params).put("id", id));
+    }
   }
 
   private void subscribeBaseListeners() {
@@ -325,14 +337,22 @@ public class DDPClientImpl {
     });
   }
 
-  private void sendMessage(String msg, @Nullable JSONBuilder json) {
+  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();
-      websocket.sendText(msg2);
+      return websocket.sendText(msg2);
     } catch (Exception e) {
       RCLog.e(e);
     }
+    return true; // ignore exception here.
+  }
+
+  private void sendMessage(String msg, @Nullable JSONBuilder json,
+                              TaskCompletionSource<?> taskForSetError) {
+    if (!sendMessage(msg, json)) {
+      taskForSetError.trySetError(new DDPClientCallback.Closed(client));
+    }
   }
 
   private void addErrorCallback(CompositeSubscription subscriptions, TaskCompletionSource<?> task) {

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

@@ -63,15 +63,15 @@ public class RxWebSocket {
     }).publish();
   }
 
-  public void sendText(String message) throws IOException {
-    webSocket.send(message);
+  public boolean sendText(String message) throws IOException {
+    return webSocket.send(message);
   }
 
   public boolean isConnected() {
     return isConnected;
   }
 
-  public void close(int code, String reason) throws IOException {
-    webSocket.close(code, reason);
+  public boolean close(int code, String reason) throws IOException {
+    return webSocket.close(code, reason);
   }
 }