Rename interface to OkHttp3SocketClientFactory.java.

ParseLiveQuery/build.gradle

@@ -41,7 +41,6 @@ android {
 
 dependencies {
     compile 'com.parse:parse-android:1.14.0'
-    compile 'com.firebase:tubesock:0.0.12'
     compile 'com.squareup.okhttp3:okhttp:3.6.0'
     compile 'com.parse.bolts:bolts-tasks:1.4.0'
 

ParseLiveQuery/src/main/java/com/parse/OkHttp3SocketClientFactory.java

@@ -0,0 +1,119 @@
+package com.parse;
+
+import android.util.Log;
+
+import java.net.URI;
+import java.util.Locale;
+
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.WebSocket;
+import okhttp3.WebSocketListener;
+import okio.ByteString;
+
+/* package */ public class OkHttp3SocketClientFactory implements WebSocketClientFactory {
+
+    OkHttpClient mClient;
+
+    public OkHttp3SocketClientFactory(OkHttpClient client) {
+        mClient = client;
+    }
+
+    public OkHttp3SocketClientFactory() {
+        mClient = new OkHttpClient();
+    }
+
+    @Override
+    public WebSocketClient createInstance(WebSocketClient.WebSocketClientCallback webSocketClientCallback, URI hostUrl) {
+        return new OkHttp3WebSocketClient(mClient, webSocketClientCallback, hostUrl);
+    }
+
+    class OkHttp3WebSocketClient implements WebSocketClient {
+
+        private static final String LOG_TAG = "OkHttpWebSocketClient";
+
+        private final WebSocketClientCallback webSocketClientCallback;
+        private WebSocket webSocket;
+        private State state = State.NONE;
+        private final OkHttpClient client;
+        private final String url;
+        private final int STATUS_CODE = 200;
+        private final String CLOSING_MSG = "User invoked close";
+
+        private final WebSocketListener handler = new WebSocketListener() {
+            @Override
+            public void onOpen(WebSocket webSocket, Response response) {
+                setState(State.CONNECTED);
+                webSocketClientCallback.onOpen();
+            }
+
+            @Override
+            public void onMessage(WebSocket webSocket, String text) {
+                webSocketClientCallback.onMessage(text);
+            }
+
+            @Override
+            public void onMessage(WebSocket webSocket, ByteString bytes) {
+                Log.w(LOG_TAG, String.format(Locale.US,
+                        "Socket got into inconsistent state and received %s instead.",
+                        bytes.toString()));
+            }
+
+            @Override
+            public void onClosed(WebSocket webSocket, int code, String reason) {
+                setState(State.DISCONNECTED);
+                webSocketClientCallback.onClose();
+            }
+
+            @Override
+            public void onFailure(okhttp3.WebSocket webSocket, Throwable t, Response response) {
+                webSocketClientCallback.onError(t);
+            }
+        };
+
+        private OkHttp3WebSocketClient(OkHttpClient okHttpClient,
+                WebSocketClientCallback webSocketClientCallback, URI hostUrl) {
+            client = okHttpClient;
+            this.webSocketClientCallback = webSocketClientCallback;
+            url = hostUrl.toString();
+        }
+
+        @Override
+        public synchronized void open() {
+            if (State.NONE == state) {
+                // OkHttp3 connects as soon as the socket is created so do it here.
+                Request request = new Request.Builder()
+                        .url(url)
+                        .build();
+
+                webSocket = client.newWebSocket(request, handler);
+                setState(State.CONNECTING);
+            }
+        }
+
+        @Override
+        public synchronized void close() {
+            setState(State.DISCONNECTING);
+            webSocket.close(STATUS_CODE, CLOSING_MSG);
+        }
+
+        @Override
+        public void send(String message) {
+            if (state == State.CONNECTED) {
+                webSocket.send(message);
+            }
+        }
+
+        @Override
+        public State getState() {
+            return state;
+        }
+
+        private synchronized void setState(State newState) {
+            this.state = newState;
+            this.webSocketClientCallback.stateChanged();
+        }
+    }
+
+}

ParseLiveQuery/src/main/java/com/parse/ParseLiveQueryClient.java

@@ -4,7 +4,6 @@
 import java.util.concurrent.Executor;
 
 public interface ParseLiveQueryClient {
-
     <T extends ParseObject> SubscriptionHandling<T> subscribe(ParseQuery<T> query);
 
     <T extends ParseObject> void unsubscribe(final ParseQuery<T> query);
@@ -21,10 +20,18 @@ public static ParseLiveQueryClient getClient() {
             return new ParseLiveQueryClientImpl();
         }
 
+        public static ParseLiveQueryClient getClient(WebSocketClientFactory webSocketClientFactory) {
+            return new ParseLiveQueryClientImpl(webSocketClientFactory);
+        }
+
         public static ParseLiveQueryClient getClient(URI uri) {
             return new ParseLiveQueryClientImpl(uri);
         }
 
+        public static ParseLiveQueryClient getClient(URI uri, WebSocketClientFactory webSocketClientFactory) {
+            return new ParseLiveQueryClientImpl(uri, webSocketClientFactory);
+        }
+
         /* package */
         static ParseLiveQueryClient getClient(URI uri, WebSocketClientFactory webSocketClientFactory, Executor taskExecutor) {
             return new ParseLiveQueryClientImpl(uri, webSocketClientFactory, taskExecutor);

ParseLiveQuery/src/main/java/com/parse/ParseLiveQueryClientImpl.java

@@ -14,6 +14,7 @@
 
 import bolts.Continuation;
 import bolts.Task;
+import okhttp3.OkHttpClient;
 
 import static com.parse.Parse.checkInit;
 
@@ -38,7 +39,15 @@
     }
 
     /* package */ ParseLiveQueryClientImpl(URI uri) {
-        this(uri, new TubeSockWebSocketClient.TubeWebSocketClientFactory(), Task.BACKGROUND_EXECUTOR);
+        this(uri, new OkHttp3SocketClientFactory(new OkHttpClient()), Task.BACKGROUND_EXECUTOR);
+    }
+
+    /* package */ ParseLiveQueryClientImpl(URI uri, WebSocketClientFactory webSocketClientFactory) {
+        this(uri, webSocketClientFactory, Task.BACKGROUND_EXECUTOR);
+    }
+
+    /* package */ ParseLiveQueryClientImpl(WebSocketClientFactory webSocketClientFactory) {
+        this(getDefaultUri(), webSocketClientFactory, Task.BACKGROUND_EXECUTOR);
     }
 
     /* package */ ParseLiveQueryClientImpl(URI uri, WebSocketClientFactory webSocketClientFactory, Executor taskExecutor) {