include android-ddp into project locally.

android-ddp/.gitignore

+/build

android-ddp/build.gradle

+apply plugin: 'com.android.library'
+apply plugin: 'me.tatarka.retrolambda'
+
+buildscript {
+  repositories {
+    jcenter()
+  }
+  dependencies {
+    classpath rootProject.ext.androidPlugin
+    classpath rootProject.ext.retroLambdaPlugin
+  }
+}
+
+android {
+  compileSdkVersion rootProject.ext.compileSdkVersion
+  buildToolsVersion rootProject.ext.buildToolsVersion
+
+  compileOptions {
+    sourceCompatibility JavaVersion.VERSION_1_8
+    targetCompatibility JavaVersion.VERSION_1_8
+  }
+  defaultConfig {
+    minSdkVersion rootProject.ext.minSdkVersion
+    targetSdkVersion rootProject.ext.compileSdkVersion
+    versionCode 1
+    versionName "0.0.8"
+  }
+  buildTypes {
+    release {
+      minifyEnabled false
+      proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
+    }
+  }
+}
+
+dependencies {
+  compile rootProject.ext.supportAnnotations
+  compile 'com.squareup.okhttp3:okhttp-ws:3.4.1'
+  compile rootProject.ext.rxJava
+  compile rootProject.ext.boltsTask
+  compile rootProject.ext.timber
+}

android-ddp/gen_listeners.py

+# -*- coding:utf-8 -*-
+
+a='''
+        @Override
+        public void onOpen(WebSocket webSocket, Response response) {
+
+        }
+
+        @Override
+        public void onFailure(IOException e, Response response) {
+
+        }
+
+        @Override
+        public void onMessage(ResponseBody responseBody) throws IOException {
+
+        }
+
+        @Override
+        public void onPong(Buffer payload) {
+
+        }
+
+        @Override
+        public void onClose(int code, String reason) {
+
+        }
+'''.strip().split('@Override')
+
+for m in a[1:]:
+    m= "        @Override\n        "+m.strip()
+    mn = m.split("\n")[1].strip().split(" ")[2].split("(")[0]
+
+    if mn.startswith("on"):
+        d=dict()
+        d["classname"]=mn[2:]
+        params = [p for p in " ".join(m.split("\n")[1].strip()[:-1].split(" throws ")[0].split(" ")[2:]).strip()[len(mn)+1:-1].split(", ") if p.split(" ")[0]!="WebSocket"]
+
+        d["params"]="".join([", "+p for p in params])
+        paramnames = [p.split(" ")[-1] for p in params]
+
+        d["paramdefs"]="\n".join(["            public "+p+";" for p in params])
+        d["thisis"]="\n".join(["                this.{param} = {param};".format(param=p) for p in paramnames])
+
+#         print '''
+#         public static class {classname} extends Base {{
+# {paramdefs}
+            
+#             public {classname}(WebSocket websocket{params}) {{
+#                 super("{classname}", websocket);
+# {thisis}
+#             }}
+#         }}'''.format(**d)
+
+        ######################
+
+        x=m.split("\n")
+        x[2]='''            mSubscriber.onNext(new RxWebSocketCallback.{classname}(mWebSocket, {params}));'''.format(classname=mn[2:],params=", ".join(paramnames))
+
+        print "\n".join(x)

android-ddp/proguard-rules.pro

+# Add project specific ProGuard rules here.
+# By default, the flags in this file are appended to flags specified
+# in /Users/yi01/Library/Android/sdk/tools/proguard/proguard-android.txt
+# You can edit the include path and order by changing the proguardFiles
+# directive in build.gradle.
+#
+# For more details, see
+#   http://developer.android.com/guide/developing/tools/proguard.html
+
+# Add any project specific keep options here:
+
+# If your project uses WebView with JS, uncomment the following
+# and specify the fully qualified class name to the JavaScript interface
+# class:
+#-keepclassmembers class fqcn.of.javascript.interface.for.webview {
+#   public *;
+#}

android-ddp/src/main/AndroidManifest.xml

+<manifest xmlns:android="http://schemas.android.com/apk/res/android"
+    package="chat.rocket.android_ddp">
+
+  <application
+      android:allowBackup="true"
+      android:label="@string/app_name"
+      android:supportsRtl="true">
+
+  </application>
+
+</manifest>

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

+package chat.rocket.android_ddp;
+
+import android.support.annotation.Nullable;
+import bolts.Task;
+import bolts.TaskCompletionSource;
+import chat.rocket.android_ddp.rx.RxWebSocketCallback;
+import okhttp3.OkHttpClient;
+import org.json.JSONArray;
+import rx.Observable;
+import timber.log.Timber;
+
+public class DDPClient {
+  // reference: https://github.com/eddflrs/meteor-ddp/blob/master/meteor-ddp.js
+
+  private final DDPClientImpl impl;
+
+  public DDPClient(OkHttpClient client) {
+    impl = new DDPClientImpl(this, client);
+    Timber.plant(new Timber.DebugTree());
+  }
+
+  public Task<DDPClientCallback.Connect> connect(String url) {
+    return connect(url, null);
+  }
+
+  public Task<DDPClientCallback.Connect> connect(String url, String session) {
+    TaskCompletionSource<DDPClientCallback.Connect> task = new TaskCompletionSource<>();
+    impl.connect(task, url, session);
+    return task.getTask();
+  }
+
+  public Task<DDPClientCallback.Ping> ping(@Nullable String id) {
+    TaskCompletionSource<DDPClientCallback.Ping> task = new TaskCompletionSource<>();
+    impl.ping(task, id);
+    return task.getTask();
+  }
+
+  public Task<DDPClientCallback.RPC> rpc(String method, JSONArray params, String id,
+      long timeoutMs) {
+    TaskCompletionSource<DDPClientCallback.RPC> task = new TaskCompletionSource<>();
+    impl.rpc(task, method, params, id, timeoutMs);
+    return task.getTask();
+  }
+
+  public Task<DDPSubscription.Ready> sub(String id, String name, JSONArray params) {
+    TaskCompletionSource<DDPSubscription.Ready> task = new TaskCompletionSource<>();
+    impl.sub(task, name, params, id);
+    return task.getTask();
+  }
+
+  public Task<DDPSubscription.NoSub> unsub(String id) {
+    TaskCompletionSource<DDPSubscription.NoSub> task = new TaskCompletionSource<>();
+    impl.unsub(task, id);
+    return task.getTask();
+  }
+
+  public Observable<DDPSubscription.Event> getSubscriptionCallback() {
+    return impl.getDDPSubscription();
+  }
+
+  public Task<RxWebSocketCallback.Close> getOnCloseCallback() {
+    return impl.getOnCloseCallback();
+  }
+
+  public boolean isConnected() {
+    return impl.isConnected();
+  }
+
+  public void close() {
+    impl.close(1000, "closed by DDPClient#close()");
+  }
+}

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

+package chat.rocket.android_ddp;
+
+import android.support.annotation.Nullable;
+import org.json.JSONObject;
+
+public class DDPClientCallback {
+  public static abstract class Base {
+    public DDPClient client;
+
+    public Base(DDPClient client) {
+      this.client = client;
+    }
+  }
+
+  public static abstract class BaseException extends Exception {
+    public DDPClient client;
+
+    public BaseException(DDPClient client) {
+      this.client = client;
+    }
+  }
+
+  public static class Connect extends Base {
+    public String session;
+
+    public Connect(DDPClient client, String session) {
+      super(client);
+      this.session = session;
+    }
+
+    public static class Failed extends BaseException {
+      public String version;
+
+      public Failed(DDPClient client, String version) {
+        super(client);
+        this.version = version;
+      }
+    }
+  }
+
+  public static class Ping extends Base {
+    @Nullable public String id;
+
+    public Ping(DDPClient client, @Nullable String id) {
+      super(client);
+      this.id = id;
+    }
+
+    public static class Timeout extends BaseException {
+      public Timeout(DDPClient client) {
+        super(client);
+      }
+    }
+  }
+
+  public static class RPC extends Base {
+    public String id;
+    public String result;
+
+    public RPC(DDPClient client, String id, String result) {
+      super(client);
+      this.id = id;
+      this.result = result;
+    }
+
+    public static class Error extends BaseException {
+      public String id;
+      public JSONObject error;
+
+      public Error(DDPClient client, String id, JSONObject error) {
+        super(client);
+        this.id = id;
+        this.error = error;
+      }
+    }
+
+    public static class Timeout extends BaseException {
+      public Timeout(DDPClient client) {
+        super(client);
+      }
+    }
+  }
+}

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

+package chat.rocket.android_ddp;
+
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+import android.text.TextUtils;
+import bolts.Task;
+import bolts.TaskCompletionSource;
+import chat.rocket.android_ddp.rx.RxWebSocket;
+import chat.rocket.android_ddp.rx.RxWebSocketCallback;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import okhttp3.OkHttpClient;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import rx.Observable;
+import rx.functions.Func1;
+import rx.subscriptions.CompositeSubscription;
+import timber.log.Timber;
+
+public class DDPClientImpl {
+  private final DDPClient client;
+  private final RxWebSocket websocket;
+  private Observable<RxWebSocketCallback.Base> observable;
+  private CompositeSubscription subscriptions;
+
+  public DDPClientImpl(DDPClient self, OkHttpClient client) {
+    websocket = new RxWebSocket(client);
+    this.client = self;
+  }
+
+  private static JSONObject toJson(String s) {
+    if (TextUtils.isEmpty(s)) return null;
+    try {
+      return new JSONObject(s);
+    } catch (JSONException e) {
+      return null;
+    }
+  }
+
+  private static String extractMsg(JSONObject response) {
+    if (response == null || response.isNull("msg")) {
+      return null;
+    } else {
+      return response.optString("msg");
+    }
+  }
+
+  public void connect(final TaskCompletionSource<DDPClientCallback.Connect> task, final String url,
+      String session) {
+    try {
+      observable = websocket.connect(url).autoConnect();
+      CompositeSubscription subscriptions = new CompositeSubscription();
+
+      subscriptions.add(observable.filter(callback -> callback instanceof RxWebSocketCallback.Open)
+          .subscribe(callback -> {
+            sendMessage("connect",
+                json -> (TextUtils.isEmpty(session) ? json : json.put("session", session)).put(
+                    "version", "pre2").put("support", new JSONArray().put("pre2").put("pre1")));
+          }, err -> {
+          }));
+
+      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 ("connected".equals(msg) && !response.isNull("session")) {
+                  task.setResult(
+                      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));
+                  subscriptions.unsubscribe();
+                } else if ("failed".equals(msg)) {
+                  task.setError(
+                      new DDPClientCallback.Connect.Failed(client, response.optString("version")));
+                  subscriptions.unsubscribe();
+                }
+              }, err -> {
+              }));
+
+      addErrorCallback(subscriptions, task);
+
+      subscribeBaseListeners();
+    } catch (Exception e) {
+      Timber.e(e);
+    }
+  }
+
+  public boolean isConnected() {
+    return websocket != null && websocket.isConnected();
+  }
+
+  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));
+                    subscriptions.unsubscribe();
+                  }
+                }
+              }
+            }, err -> {
+              task.setError(new DDPClientCallback.Ping.Timeout(client));
+            }));
+
+    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) {
+    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.setError(new DDPSubscription.NoSub.Error(client, id,
+                      response.optJSONObject("error")));
+                  subscriptions.unsubscribe();
+                }
+              }
+            }, err -> {
+            }));
+
+    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) {
+    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 -> {
+            }));
+
+    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) {
+    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();
+                }
+              }
+            }, err -> {
+              if (err instanceof TimeoutException) {
+                task.setError(new DDPClientCallback.RPC.Timeout(client));
+              }
+            }));
+
+    addErrorCallback(subscriptions, task);
+
+    sendMessage("method", json -> json.put("method", method).put("params", params).put("id", id));
+  }
+
+  private void subscribeBaseListeners() {
+    if (subscriptions != null &&
+        subscriptions.hasSubscriptions() && !subscriptions.isUnsubscribed()) {
+      return;
+    }
+
+    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 ("ping".equals(msg)) {
+                if (response.isNull("id")) {
+                  sendMessage("pong", null);
+                } else {
+                  sendMessage("pong", json -> json.put("id", response.getString("id")));
+                }
+              }
+            }, err -> {
+            }));
+  }
+
+  public Observable<DDPSubscription.Event> getDDPSubscription() {
+    String[] targetMsgs = { "added", "changed", "removed", "addedBefore", "movedBefore" };
+    return observable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
+        .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
+        .map(DDPClientImpl::toJson)
+        .filter(response -> {
+          String msg = extractMsg(response);
+          for (String m : targetMsgs) {
+            if (m.equals(msg)) return true;
+          }
+          return false;
+        })
+        .map((Func1<JSONObject, DDPSubscription.Event>) response -> {
+          String msg = extractMsg(response);
+          if ("added".equals(msg)) {
+            return new DDPSubscription.Added(client, response.optString("collection"),
+                response.optString("id"),
+                response.isNull("fields") ? null : response.optJSONObject("fields"));
+          } else if ("addedBefore".equals(msg)) {
+            return new DDPSubscription.Added.Before(client, response.optString("collection"),
+                response.optString("id"),
+                response.isNull("fields") ? null : response.optJSONObject("fields"),
+                response.isNull("before") ? null : response.optString("before"));
+          } else if ("changed".equals(msg)) {
+            return new DDPSubscription.Changed(client, response.optString("collection"),
+                response.optString("id"),
+                response.isNull("fields") ? null : response.optJSONObject("fields"),
+                response.isNull("cleared") ? new JSONArray() : response.optJSONArray("before"));
+          } else if ("removed".equals(msg)) {
+            return new DDPSubscription.Removed(client, response.optString("collection"),
+                response.optString("id"));
+          } else if ("movedBefore".equals(msg)) {
+            return new DDPSubscription.MovedBefore(client, response.optString("collection"),
+                response.optString("id"),
+                response.isNull("before") ? null : response.optString("before"));
+          }
+
+          return null;
+        })
+        .asObservable();
+  }
+
+  public void unsubscribeBaseListeners() {
+    if (subscriptions.hasSubscriptions() && !subscriptions.isUnsubscribed()) {
+      subscriptions.unsubscribe();
+    }
+  }
+
+  public Task<RxWebSocketCallback.Close> getOnCloseCallback() {
+    TaskCompletionSource<RxWebSocketCallback.Close> task = new TaskCompletionSource<>();
+
+    observable.filter(callback -> callback instanceof RxWebSocketCallback.Close)
+        .cast(RxWebSocketCallback.Close.class)
+        .subscribe(close -> {
+          task.setResult(close);
+        }, err -> {
+          if (err instanceof Exception) {
+            task.setError((Exception) err);
+          } else {
+            task.setError(new Exception(err));
+          }
+        });
+
+    return task.getTask().onSuccessTask(_task -> {
+      unsubscribeBaseListeners();
+      return _task;
+    });
+  }
+
+  private void 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);
+    } catch (Exception e) {
+      Timber.e(e);
+    }
+  }
+
+  private void addErrorCallback(CompositeSubscription subscriptions, TaskCompletionSource<?> task) {
+    subscriptions.add(observable.subscribe(base -> {
+    }, err -> {
+      task.setError(new Exception(err));
+      subscriptions.unsubscribe();
+    }));
+  }
+
+  public void close(int code, String reason) {
+    try {
+      websocket.close(code, reason);
+    } catch (Exception e) {
+      Timber.e(e);
+    }
+  }
+
+  private interface JSONBuilder {
+    @NonNull JSONObject create(JSONObject root) throws JSONException;
+  }
+}

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

+package chat.rocket.android_ddp;
+
+import android.support.annotation.NonNull;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+public class DDPSubscription {
+  public static abstract class Event {
+    public final DDPClient client;
+
+    public Event(DDPClient client) {
+      this.client = client;
+    }
+  }
+
+  public static abstract class BaseException extends Exception {
+    public final DDPClient client;
+
+    public BaseException(DDPClient client) {
+      this.client = client;
+    }
+  }
+
+  public static class NoSub extends Event {
+    public String id;
+
+    public NoSub(DDPClient client, String id) {
+      super(client);
+      this.id = id;
+    }
+
+    @Override public String toString() {
+      return "NoSub[id=" + id + "]";
+    }
+
+    public static class Error extends BaseException {
+      String id;
+      JSONObject error;
+
+      public Error(DDPClient client, String id, JSONObject error) {
+        super(client);
+        this.id = id;
+        this.error = error;
+      }
+    }
+  }
+
+  public static class Ready extends Event {
+    public String id;
+
+    public Ready(DDPClient client, String id) {
+      super(client);
+      this.id = id;
+    }
+
+    @Override public String toString() {
+      return "Ready[id=" + id + "]";
+    }
+  }
+
+  public static class DocEvent extends Event {
+    public String collection;
+    public String docID;
+
+    public DocEvent(DDPClient client, String collection, String docID) {
+      super(client);
+      this.collection = collection;
+      this.docID = docID;
+    }
+
+    @Override public String toString() {
+      return "DocEvent[id=" + docID + ", collection=" + collection + "]";
+    }
+  }
+
+  public static class Added extends DocEvent {
+    public JSONObject fields;
+
+    public Added(DDPClient client, String collection, String docID, JSONObject fields) {
+      super(client, collection, docID);
+      this.fields = fields;
+    }
+
+    public static class Before extends Added {
+      public String before;
+
+      public Before(DDPClient client, String collection, String docID, JSONObject fields,
+          String before) {
+        super(client, collection, docID, fields);
+        this.before = before;
+      }
+    }
+  }
+
+  public static class Changed extends DocEvent {
+    public JSONObject fields;
+    public JSONArray cleared;
+
+    public Changed(DDPClient client, String collection, String docID, JSONObject fields,
+        @NonNull JSONArray cleared) {
+      super(client, collection, docID);
+      this.fields = fields;
+      this.cleared = cleared;
+    }
+  }
+
+  public static class Removed extends DocEvent {
+
+    public Removed(DDPClient client, String collection, String docID) {
+      super(client, collection, docID);
+    }
+  }
+
+  public static class MovedBefore extends DocEvent {
+    public String before;
+
+    public MovedBefore(DDPClient client, String collection, String docID, String before) {
+      super(client, collection, docID);
+      this.before = before;
+    }
+  }
+}

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

+package chat.rocket.android_ddp.rx;
+
+import java.io.IOException;
+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 rx.Observable;
+import rx.Subscriber;
+import rx.exceptions.OnErrorNotImplementedException;
+import rx.observables.ConnectableObservable;
+import timber.log.Timber;
+
+public class RxWebSocket {
+  private OkHttpClient httpClient;
+  private WebSocket webSocket;
+  private boolean isConnected;
+
+  public RxWebSocket(OkHttpClient client) {
+    httpClient = client;
+    isConnected = false;
+  }
+
+  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() {
+          @Override public void onOpen(WebSocket webSocket, Response response) {
+            isConnected = true;
+            RxWebSocket.this.webSocket = webSocket;
+            subscriber.onNext(new RxWebSocketCallback.Open(RxWebSocket.this.webSocket, response));
+          }
+
+          @Override public void onFailure(IOException e, Response response) {
+            try {
+              isConnected = false;
+              subscriber.onError(new RxWebSocketCallback.Failure(webSocket, e, response));
+            } catch (OnErrorNotImplementedException ex) {
+              Timber.w(ex, "OnErrorNotImplementedException ignored");
+            }
+          }
+
+          @Override public void onMessage(ResponseBody responseBody) throws IOException {
+            isConnected = true;
+            subscriber.onNext(new RxWebSocketCallback.Message(webSocket, responseBody));
+          }
+
+          @Override public void onPong(Buffer payload) {
+            isConnected = true;
+            subscriber.onNext(new RxWebSocketCallback.Pong(webSocket, payload));
+          }
+
+          @Override public void onClose(int code, String reason) {
+            isConnected = false;
+            subscriber.onNext(new RxWebSocketCallback.Close(webSocket, code, reason));
+            subscriber.onCompleted();
+          }
+        });
+      }
+    }).publish();
+  }
+
+  public void sendText(String message) throws IOException {
+    webSocket.sendMessage(RequestBody.create(WebSocket.TEXT, message));
+  }
+
+  public boolean isConnected() {
+    return isConnected;
+  }
+
+  public void close(int code, String reason) throws IOException {
+    webSocket.close(code, reason);
+  }
+}

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

+package chat.rocket.android_ddp.rx;
+
+import java.io.IOException;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okhttp3.ws.WebSocket;
+import okio.Buffer;
+import timber.log.Timber;
+
+import static android.R.attr.type;
+
+public class RxWebSocketCallback {
+  public static abstract class Base {
+    public String type;
+    public WebSocket ws;
+
+    public Base(String type, WebSocket ws) {
+      this.type = type;
+      this.ws = ws;
+    }
+
+    @Override public String toString() {
+      return "[" + type + "]";
+    }
+  }
+
+  public static class Open extends Base {
+    public Response response;
+
+    public Open(WebSocket websocket, Response response) {
+      super("Open", websocket);
+      this.response = response;
+    }
+  }
+
+  public static class Failure extends Exception {
+    public WebSocket ws;
+    public Response response;
+
+    public Failure(WebSocket websocket, IOException e, Response response) {
+      super(e);
+      this.ws = websocket;
+      this.response = response;
+    }
+
+    @Override public String toString() {
+      if (response != null) {
+        return "[" + type + "] " + response.message();
+      } else {
+        return super.toString();
+      }
+    }
+  }
+
+  public static class Message extends Base {
+    public String responseBodyString;
+
+    public Message(WebSocket websocket, ResponseBody responseBody) {
+      super("Message", websocket);
+      try {
+        this.responseBodyString = responseBody.string();
+      } catch (Exception e) {
+        Timber.e(e, "error in reading response(Message)");
+      }
+    }
+
+    @Override public String toString() {
+      return "[" + type + "] " + responseBodyString;
+    }
+  }
+
+  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;
+
+    public Close(WebSocket websocket, int code, String reason) {
+      super("Close", websocket);
+      this.code = code;
+      this.reason = reason;
+    }
+
+    @Override public String toString() {
+      return "[" + type + "] code=" + code + ", reason=" + reason;
+    }
+  }
+}

android-ddp/src/main/res/values/strings.xml

+<resources>
+  <string name="app_name">android-ddp</string>
+</resources>

app/build.gradle

@@ -14,7 +14,7 @@ buildscript {
 
     // NOTE: Do not place your application dependencies here; they belong
     // in the individual module build.gradle files
-    classpath 'me.tatarka:gradle-retrolambda:3.3.1'
+    classpath rootProject.ext.retroLambdaPlugin
     classpath 'me.tatarka.retrolambda.projectlombok:lombok.ast:0.2.3.a2'
     classpath rootProject.ext.realmPlugin
     classpath 'com.jakewharton.hugo:hugo-plugin:1.2.1'
@@ -78,6 +78,7 @@ repositories {
 }
 
 dependencies {
+  compile project(':android-ddp')
   compile project(':rocket-chat-android-widgets')
   compile project(':realm-helpers')
   compile rootProject.ext.supportAppCompat
@@ -98,8 +99,6 @@ dependencies {
   compile 'com.facebook.stetho:stetho-okhttp3:1.4.1'
   compile 'com.uphyca:stetho_realm:2.0.1'
 
-  compile 'chat.rocket:android-ddp:0.0.8'
-
   compile rootProject.ext.timber
   compile 'com.jakewharton.rxbinding:rxbinding:0.4.0'
   compile 'com.jakewharton.rxbinding:rxbinding-support-v4:0.4.0'

dependencies.gradle

 ext {
   androidPlugin = 'com.android.tools.build:gradle:2.2.2'
   realmPlugin = 'io.realm:realm-gradle-plugin:2.2.1'
+  retroLambdaPlugin = 'me.tatarka:gradle-retrolambda:3.3.1'
   compileSdkVersion = 25
   buildToolsVersion = '25.0.1'
   minSdkVersion = 21 //for accelerating multi-dex build. OVERRIDEN BY Circle CI to 17

settings.gradle

-include ':app', ':rocket-chat-android-widgets', ':realm-helpers'
+include ':app', ':android-ddp', ':rocket-chat-android-widgets', ':realm-helpers'