Merge branch 'origin/okhttp3.5.0' into develop

android-ddp/build.gradle

@@ -37,7 +37,7 @@ android {
 dependencies {
   compile project(':log-wrapper')
   compile rootProject.ext.supportAnnotations
-  compile 'com.squareup.okhttp3:okhttp-ws:3.4.1'
+  compile rootProject.ext.okhttp3
   compile rootProject.ext.rxJava
   compile rootProject.ext.boltsTask
 }

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,135 +102,146 @@ public class DDPClientImpl {
 
   public void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
                    @Nullable final String id) {
-    CompositeSubscription subscriptions = new CompositeSubscription();
 
-    subscriptions.add(
-        observable.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);
-              if ("pong".equals(msg)) {
-                if (response.isNull("id")) {
-                  task.setResult(new DDPClientCallback.Ping(client, null));
-                  subscriptions.unsubscribe();
-                } else {
-                  String _id = response.optString("id");
-                  if (id.equals(_id)) {
-                    task.setResult(new DDPClientCallback.Ping(client, 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(
+          observable.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);
+                if ("pong".equals(msg)) {
+                  if (response.isNull("id")) {
+                    task.setResult(new DDPClientCallback.Ping(client, null));
                     subscriptions.unsubscribe();
+                  } else {
+                    String _id = response.optString("id");
+                    if (id.equals(_id)) {
+                      task.setResult(new DDPClientCallback.Ping(client, id));
+                      subscriptions.unsubscribe();
+                    }
                   }
                 }
-              }
-            }, err -> {
-              task.setError(new DDPClientCallback.Ping.Timeout(client));
-            }));
-
-    addErrorCallback(subscriptions, task);
+              }, err -> {
+                task.setError(new DDPClientCallback.Ping.Timeout(client));
+              }));
 
-    if (TextUtils.isEmpty(id)) {
-      sendMessage("ping", null);
-    } else {
-      sendMessage("ping", json -> json.put("id", id));
+      addErrorCallback(subscriptions, task);
     }
   }
 
   public void sub(final TaskCompletionSource<DDPSubscription.Ready> task, String name,
                   JSONArray params, String id) {
-    CompositeSubscription subscriptions = new CompositeSubscription();
+    final boolean requested =
+        sendMessage("sub", json -> json.put("id", id).put("name", name).put("params", params));
 
-    subscriptions.add(
-        observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
-            .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
-            .map(DDPClientImpl::toJson)
-            .subscribe(response -> {
-              String msg = extractMsg(response);
-              if ("ready".equals(msg) && !response.isNull("subs")) {
-                JSONArray ids = response.optJSONArray("subs");
-                for (int i = 0; i < ids.length(); i++) {
-                  String _id = ids.optString(i);
+    if (requested) {
+      CompositeSubscription subscriptions = new CompositeSubscription();
+
+      subscriptions.add(
+          observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
+              .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
+              .map(DDPClientImpl::toJson)
+              .subscribe(response -> {
+                String msg = extractMsg(response);
+                if ("ready".equals(msg) && !response.isNull("subs")) {
+                  JSONArray ids = response.optJSONArray("subs");
+                  for (int i = 0; i < ids.length(); i++) {
+                    String _id = ids.optString(i);
+                    if (id.equals(_id)) {
+                      task.setResult(new DDPSubscription.Ready(client, id));
+                      subscriptions.unsubscribe();
+                      break;
+                    }
+                  }
+                } else if ("nosub".equals(msg) && !response.isNull("id") && !response.isNull(
+                    "error")) {
+                  String _id = response.optString("id");
                   if (id.equals(_id)) {
-                    task.setResult(new DDPSubscription.Ready(client, id));
+                    task.setError(new DDPSubscription.NoSub.Error(client, id,
+                        response.optJSONObject("error")));
                     subscriptions.unsubscribe();
-                    break;
                   }
                 }
-              } else if ("nosub".equals(msg) && !response.isNull("id") && !response.isNull(
-                  "error")) {
-                String _id = response.optString("id");
-                if (id.equals(_id)) {
-                  task.setError(new DDPSubscription.NoSub.Error(client, id,
-                      response.optJSONObject("error")));
-                  subscriptions.unsubscribe();
-                }
-              }
-            }, err -> {
-            }));
-
-    addErrorCallback(subscriptions, task);
+              }, err -> {
+              }));
 
-    sendMessage("sub", json -> json.put("id", id).put("name", name).put("params", params));
+      addErrorCallback(subscriptions, task);
+    }
   }
 
   public void unsub(final TaskCompletionSource<DDPSubscription.NoSub> task,
                     @Nullable final String id) {
-    CompositeSubscription subscriptions = new CompositeSubscription();
 
-    subscriptions.add(
-        observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
-            .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
-            .map(DDPClientImpl::toJson)
-            .subscribe(response -> {
-              String msg = extractMsg(response);
-              if ("nosub".equals(msg) && response.isNull("error") && !response.isNull("id")) {
-                String _id = response.optString("id");
-                if (id.equals(_id)) {
-                  task.setResult(new DDPSubscription.NoSub(client, id));
-                  subscriptions.unsubscribe();
-                }
-              }
-            }, err -> {
-            }));
+    final boolean requested = sendMessage("unsub", json -> json.put("id", id));
 
-    addErrorCallback(subscriptions, task);
+    if (requested) {
+      CompositeSubscription subscriptions = new CompositeSubscription();
 
-    sendMessage("unsub", json -> json.put("id", id));
+      subscriptions.add(
+          observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
+              .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
+              .map(DDPClientImpl::toJson)
+              .subscribe(response -> {
+                String msg = extractMsg(response);
+                if ("nosub".equals(msg) && response.isNull("error") && !response.isNull("id")) {
+                  String _id = response.optString("id");
+                  if (id.equals(_id)) {
+                    task.setResult(new DDPSubscription.NoSub(client, id));
+                    subscriptions.unsubscribe();
+                  }
+                }
+              }, err -> {
+              }));
+
+      addErrorCallback(subscriptions, task);
+    }
   }
 
   public void rpc(final TaskCompletionSource<DDPClientCallback.RPC> task, String method,
                   JSONArray params, String id, long timeoutMs) {
-    CompositeSubscription subscriptions = new CompositeSubscription();
+    final boolean requested =
+        sendMessage("method",
+            json -> json.put("method", method).put("params", params).put("id", id));
 
-    subscriptions.add(
-        observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
-            .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
-            .map(DDPClientImpl::toJson)
-            .timeout(timeoutMs, TimeUnit.MILLISECONDS)
-            .subscribe(response -> {
-              String msg = extractMsg(response);
-              if ("result".equals(msg)) {
-                String _id = response.optString("id");
-                if (id.equals(_id)) {
-                  if (!response.isNull("error")) {
-                    task.setError(new DDPClientCallback.RPC.Error(client, id,
-                        response.optJSONObject("error")));
-                  } else {
-                    String result = response.optString("result");
-                    task.setResult(new DDPClientCallback.RPC(client, id, result));
+    if (requested) {
+      CompositeSubscription subscriptions = new CompositeSubscription();
+
+      subscriptions.add(
+          observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
+              .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
+              .map(DDPClientImpl::toJson)
+              .timeout(timeoutMs, TimeUnit.MILLISECONDS)
+              .subscribe(response -> {
+                String msg = extractMsg(response);
+                if ("result".equals(msg)) {
+                  String _id = response.optString("id");
+                  if (id.equals(_id)) {
+                    if (!response.isNull("error")) {
+                      task.setError(new DDPClientCallback.RPC.Error(client, id,
+                          response.optJSONObject("error")));
+                    } else {
+                      String result = response.optString("result");
+                      task.setResult(new DDPClientCallback.RPC(client, id, result));
+                    }
+                    subscriptions.unsubscribe();
                   }
-                  subscriptions.unsubscribe();
                 }
-              }
-            }, err -> {
-              if (err instanceof TimeoutException) {
-                task.setError(new DDPClientCallback.RPC.Timeout(client));
-              }
-            }));
-
-    addErrorCallback(subscriptions, task);
+              }, err -> {
+                if (err instanceof TimeoutException) {
+                  task.setError(new DDPClientCallback.RPC.Timeout(client));
+                }
+              }));
 
-    sendMessage("method", json -> json.put("method", method).put("params", params).put("id", id));
+      addErrorCallback(subscriptions, task);
+    }
   }
 
   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

@@ -4,13 +4,9 @@ import java.io.IOException;
 import chat.rocket.android.log.RCLog;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
-import okhttp3.RequestBody;
 import okhttp3.Response;
-import okhttp3.ResponseBody;
-import okhttp3.ws.WebSocket;
-import okhttp3.ws.WebSocketCall;
-import okhttp3.ws.WebSocketListener;
-import okio.Buffer;
+import okhttp3.WebSocket;
+import okhttp3.WebSocketListener;
 import rx.Observable;
 import rx.Subscriber;
 import rx.exceptions.OnErrorNotImplementedException;
@@ -28,12 +24,11 @@ public class RxWebSocket {
 
   public ConnectableObservable<RxWebSocketCallback.Base> connect(String url) {
     final Request request = new Request.Builder().url(url).build();
-    WebSocketCall call = WebSocketCall.create(httpClient, request);
 
     return Observable.create(new Observable.OnSubscribe<RxWebSocketCallback.Base>() {
       @Override
       public void call(Subscriber<? super RxWebSocketCallback.Base> subscriber) {
-        call.enqueue(new WebSocketListener() {
+        httpClient.newWebSocket(request, new WebSocketListener() {
           @Override
           public void onOpen(WebSocket webSocket, Response response) {
             isConnected = true;
@@ -42,29 +37,23 @@ public class RxWebSocket {
           }
 
           @Override
-          public void onFailure(IOException e, Response response) {
+          public void onFailure(WebSocket webSocket, Throwable err, Response response) {
             try {
               isConnected = false;
-              subscriber.onError(new RxWebSocketCallback.Failure(webSocket, e, response));
+              subscriber.onError(new RxWebSocketCallback.Failure(webSocket, err, response));
             } catch (OnErrorNotImplementedException ex) {
               RCLog.w(ex, "OnErrorNotImplementedException ignored");
             }
           }
 
           @Override
-          public void onMessage(ResponseBody responseBody) throws IOException {
+          public void onMessage(WebSocket webSocket, String text) {
             isConnected = true;
-            subscriber.onNext(new RxWebSocketCallback.Message(webSocket, responseBody));
+            subscriber.onNext(new RxWebSocketCallback.Message(webSocket, text));
           }
 
           @Override
-          public void onPong(Buffer payload) {
-            isConnected = true;
-            subscriber.onNext(new RxWebSocketCallback.Pong(webSocket, payload));
-          }
-
-          @Override
-          public void onClose(int code, String reason) {
+          public void onClosed(WebSocket webSocket, int code, String reason) {
             isConnected = false;
             subscriber.onNext(new RxWebSocketCallback.Close(webSocket, code, reason));
             subscriber.onCompleted();
@@ -74,15 +63,15 @@ public class RxWebSocket {
     }).publish();
   }
 
-  public void sendText(String message) throws IOException {
-    webSocket.sendMessage(RequestBody.create(WebSocket.TEXT, 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);
   }
 }

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

@@ -2,12 +2,9 @@ package chat.rocket.android_ddp.rx;
 
 import static android.R.attr.type;
 
-import java.io.IOException;
 import chat.rocket.android.log.RCLog;
 import okhttp3.Response;
-import okhttp3.ResponseBody;
-import okhttp3.ws.WebSocket;
-import okio.Buffer;
+import okhttp3.WebSocket;
 
 public class RxWebSocketCallback {
   public static abstract class Base {
@@ -38,8 +35,8 @@ public class RxWebSocketCallback {
     public WebSocket ws;
     public Response response;
 
-    public Failure(WebSocket websocket, IOException e, Response response) {
-      super(e);
+    public Failure(WebSocket websocket, Throwable err, Response response) {
+      super(err);
       this.ws = websocket;
       this.response = response;
     }
@@ -57,10 +54,10 @@ public class RxWebSocketCallback {
   public static class Message extends Base {
     public String responseBodyString;
 
-    public Message(WebSocket websocket, ResponseBody responseBody) {
+    public Message(WebSocket websocket, String responseBody) {
       super("Message", websocket);
       try {
-        this.responseBodyString = responseBody.string();
+        this.responseBodyString = responseBody;
       } catch (Exception e) {
         RCLog.e(e, "error in reading response(Message)");
       }
@@ -72,15 +69,6 @@ public class RxWebSocketCallback {
     }
   }
 
-  public static class Pong extends Base {
-    public Buffer payload;
-
-    public Pong(WebSocket websocket, Buffer payload) {
-      super("Pong", websocket);
-      this.payload = payload;
-    }
-  }
-
   public static class Close extends Base {
     public int code;
     public String reason;

dependencies.gradle

@@ -15,7 +15,7 @@ ext {
 
   rxJava = 'io.reactivex:rxjava:1.2.2'
   boltsTask = 'com.parse.bolts:bolts-tasks:1.4.0'
-  okhttp3 = 'com.squareup.okhttp3:okhttp:3.4.1'
+  okhttp3 = 'com.squareup.okhttp3:okhttp:3.5.0'
   picasso = 'com.squareup.picasso:picasso:2.5.2'
   picasso2Okhttp3Downloader = 'com.jakewharton.picasso:picasso2-okhttp3-downloader:1.1.0'
   textDrawable = 'com.amulyakhare:com.amulyakhare.textdrawable:1.0.1'