Support for retrieving stored messages via websocket.

1) When registering with server, indicate that the server should
   store messages and send notifications.

2) Process notification GCM messages, and connect to the server
   to retrieve actual message content.

libtextsecure/src/main/java/org/whispersystems/textsecure/api/TextSecureAccountManager.java

@@ -30,6 +30,7 @@ import org.whispersystems.textsecure.api.push.SignedPreKeyEntity;
 import org.whispersystems.textsecure.api.push.TrustStore;
 import org.whispersystems.textsecure.internal.crypto.ProvisioningCipher;
 import org.whispersystems.textsecure.internal.push.PushServiceSocket;
+import org.whispersystems.textsecure.internal.util.StaticCredentialsProvider;
 
 import java.io.IOException;
 import java.util.List;
@@ -45,7 +46,7 @@ public class TextSecureAccountManager {
   public TextSecureAccountManager(String url, TrustStore trustStore,
                                   String user, String password)
   {
-    this.pushServiceSocket = new PushServiceSocket(url, trustStore, user, password);
+    this.pushServiceSocket = new PushServiceSocket(url, trustStore, new StaticCredentialsProvider(user, password, null));
     this.user              = user;
   }
 

libtextsecure/src/main/java/org/whispersystems/textsecure/api/TextSecureMessagePipe.java

@@ -0,0 +1,87 @@
+package org.whispersystems.textsecure.api;
+
+import org.whispersystems.libaxolotl.InvalidVersionException;
+import org.whispersystems.textsecure.api.messages.TextSecureEnvelope;
+import org.whispersystems.textsecure.api.util.CredentialsProvider;
+import org.whispersystems.textsecure.internal.websocket.WebSocketConnection;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.whispersystems.textsecure.internal.websocket.WebSocketProtos.WebSocketRequestMessage;
+import static org.whispersystems.textsecure.internal.websocket.WebSocketProtos.WebSocketResponseMessage;
+
+public class TextSecureMessagePipe {
+
+  private final WebSocketConnection websocket;
+  private final CredentialsProvider credentialsProvider;
+
+  public TextSecureMessagePipe(WebSocketConnection websocket, CredentialsProvider credentialsProvider) {
+    this.websocket           = websocket;
+    this.credentialsProvider = credentialsProvider;
+
+    this.websocket.connect();
+  }
+
+  public TextSecureEnvelope read(long timeout, TimeUnit unit)
+      throws InvalidVersionException, IOException, TimeoutException
+  {
+    return read(timeout, unit, new NullMessagePipeCallback());
+  }
+
+  public TextSecureEnvelope read(long timeout, TimeUnit unit, MessagePipeCallback callback)
+      throws TimeoutException, IOException, InvalidVersionException
+  {
+    while (true) {
+      WebSocketRequestMessage  request  = websocket.readRequest(unit.toMillis(timeout));
+      WebSocketResponseMessage response = createWebSocketResponse(request);
+
+      try {
+        if (isTextSecureEnvelope(request)) {
+          TextSecureEnvelope envelope = new TextSecureEnvelope(request.getBody().toByteArray(),
+                                                               credentialsProvider.getSignalingKey());
+
+          callback.onMessage(envelope);
+          return envelope;
+        }
+      } finally {
+        websocket.sendResponse(response);
+      }
+    }
+  }
+
+  public void shutdown() throws IOException {
+    websocket.disconnect();
+  }
+
+  private boolean isTextSecureEnvelope(WebSocketRequestMessage message) {
+    return "PUT".equals(message.getVerb()) && "/api/v1/message".equals(message.getPath());
+  }
+
+  private WebSocketResponseMessage createWebSocketResponse(WebSocketRequestMessage request) {
+    if (isTextSecureEnvelope(request)) {
+      return WebSocketResponseMessage.newBuilder()
+                                     .setId(request.getId())
+                                     .setStatus(200)
+                                     .setMessage("OK")
+                                     .build();
+    } else {
+      return WebSocketResponseMessage.newBuilder()
+                                     .setId(request.getId())
+                                     .setStatus(400)
+                                     .setMessage("Unknown")
+                                     .build();
+    }
+  }
+
+  public static interface MessagePipeCallback {
+    public void onMessage(TextSecureEnvelope envelope);
+  }
+
+  private static class NullMessagePipeCallback implements MessagePipeCallback {
+    @Override
+    public void onMessage(TextSecureEnvelope envelope) {}
+  }
+
+}

libtextsecure/src/main/java/org/whispersystems/textsecure/api/TextSecureMessageReceiver.java

@@ -17,23 +17,43 @@
 package org.whispersystems.textsecure.api;
 
 import org.whispersystems.libaxolotl.InvalidMessageException;
+import org.whispersystems.libaxolotl.InvalidVersionException;
 import org.whispersystems.textsecure.api.crypto.AttachmentCipherInputStream;
 import org.whispersystems.textsecure.api.messages.TextSecureAttachmentPointer;
+import org.whispersystems.textsecure.api.messages.TextSecureEnvelope;
 import org.whispersystems.textsecure.api.push.TrustStore;
+import org.whispersystems.textsecure.api.util.CredentialsProvider;
 import org.whispersystems.textsecure.internal.push.PushServiceSocket;
+import org.whispersystems.textsecure.internal.util.StaticCredentialsProvider;
+import org.whispersystems.textsecure.internal.websocket.WebSocketConnection;
+import org.whispersystems.textsecure.internal.websocket.WebSocketProtos;
 
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.whispersystems.textsecure.internal.websocket.WebSocketProtos.WebSocketRequestMessage;
 
 public class TextSecureMessageReceiver {
 
-  private final PushServiceSocket socket;
+  private final PushServiceSocket   socket;
+  private final TrustStore          trustStore;
+  private final String              url;
+  private final CredentialsProvider credentialsProvider;
 
   public TextSecureMessageReceiver(String url, TrustStore trustStore,
-                                   String user, String password)
+                                   String user, String password, String signalingKey)
   {
-    this.socket = new PushServiceSocket(url, trustStore, user, password);
+    this(url, trustStore, new StaticCredentialsProvider(user, password, signalingKey));
+  }
+
+  public TextSecureMessageReceiver(String url, TrustStore trustStore, CredentialsProvider credentials) {
+    this.url                 = url;
+    this.trustStore          = trustStore;
+    this.credentialsProvider = credentials;
+    this.socket              = new PushServiceSocket(url, trustStore, credentials);
   }
 
   public InputStream retrieveAttachment(TextSecureAttachmentPointer pointer, File destination)
@@ -43,4 +63,9 @@ public class TextSecureMessageReceiver {
     return new AttachmentCipherInputStream(destination, pointer.getKey());
   }
 
+  public TextSecureMessagePipe createMessagePipe() {
+    WebSocketConnection webSocket = new WebSocketConnection(url, trustStore, credentialsProvider);
+    return new TextSecureMessagePipe(webSocket, credentialsProvider);
+  }
+
 }

libtextsecure/src/main/java/org/whispersystems/textsecure/api/TextSecureMessageSender.java

@@ -47,6 +47,7 @@ import org.whispersystems.textsecure.api.push.exceptions.UnregisteredUserExcepti
 import org.whispersystems.textsecure.api.push.exceptions.EncapsulatedExceptions;
 import org.whispersystems.textsecure.internal.push.exceptions.MismatchedDevicesException;
 import org.whispersystems.textsecure.internal.push.exceptions.StaleDevicesException;
+import org.whispersystems.textsecure.internal.util.StaticCredentialsProvider;
 import org.whispersystems.textsecure.internal.util.Util;
 
 import java.io.IOException;
@@ -72,7 +73,7 @@ public class TextSecureMessageSender {
                                  long userId, AxolotlStore store,
                                  Optional<EventListener> eventListener)
   {
-    this.socket        = new PushServiceSocket(url, trustStore, user, password);
+    this.socket        = new PushServiceSocket(url, trustStore, new StaticCredentialsProvider(user, password, null));
     this.store         = store;
     this.syncAddress   = new PushAddress(userId, user, null);
     this.eventListener = eventListener;

libtextsecure/src/main/java/org/whispersystems/textsecure/api/messages/TextSecureEnvelope.java

@@ -59,8 +59,12 @@ public class TextSecureEnvelope {
   public TextSecureEnvelope(String message, String signalingKey)
       throws IOException, InvalidVersionException
   {
-    byte[] ciphertext = Base64.decode(message);
+    this(Base64.decode(message), signalingKey);
+  }
 
+  public TextSecureEnvelope(byte[] ciphertext, String signalingKey)
+      throws InvalidVersionException, IOException
+  {
     if (ciphertext.length < VERSION_LENGTH || ciphertext[VERSION_OFFSET] != SUPPORTED_VERSION)
       throw new InvalidVersionException("Unsupported version!");
 

libtextsecure/src/main/java/org/whispersystems/textsecure/api/util/CredentialsProvider.java

@@ -0,0 +1,7 @@
+package org.whispersystems.textsecure.api.util;
+
+public interface CredentialsProvider {
+  public String getUser();
+  public String getPassword();
+  public String getSignalingKey();
+}

libtextsecure/src/main/java/org/whispersystems/textsecure/internal/push/PushServiceSocket.java

@@ -27,23 +27,24 @@ import org.whispersystems.libaxolotl.ecc.ECPublicKey;
 import org.whispersystems.libaxolotl.state.PreKeyBundle;
 import org.whispersystems.libaxolotl.state.PreKeyRecord;
 import org.whispersystems.libaxolotl.state.SignedPreKeyRecord;
-import org.whispersystems.textsecure.api.push.PushAddress;
 import org.whispersystems.textsecure.api.crypto.AttachmentCipherOutputStream;
 import org.whispersystems.textsecure.api.push.ContactTokenDetails;
+import org.whispersystems.textsecure.api.push.PushAddress;
 import org.whispersystems.textsecure.api.push.SignedPreKeyEntity;
 import org.whispersystems.textsecure.api.push.TrustStore;
-import org.whispersystems.textsecure.api.push.exceptions.UnregisteredUserException;
-import org.whispersystems.textsecure.internal.push.exceptions.MismatchedDevicesException;
-import org.whispersystems.textsecure.internal.push.exceptions.StaleDevicesException;
-import org.whispersystems.textsecure.internal.util.Base64;
-import org.whispersystems.textsecure.internal.util.Util;
 import org.whispersystems.textsecure.api.push.exceptions.AuthorizationFailedException;
 import org.whispersystems.textsecure.api.push.exceptions.ExpectationFailedException;
 import org.whispersystems.textsecure.api.push.exceptions.NonSuccessfulResponseCodeException;
 import org.whispersystems.textsecure.api.push.exceptions.NotFoundException;
 import org.whispersystems.textsecure.api.push.exceptions.PushNetworkException;
 import org.whispersystems.textsecure.api.push.exceptions.RateLimitException;
+import org.whispersystems.textsecure.api.push.exceptions.UnregisteredUserException;
+import org.whispersystems.textsecure.api.util.CredentialsProvider;
+import org.whispersystems.textsecure.internal.push.exceptions.MismatchedDevicesException;
+import org.whispersystems.textsecure.internal.push.exceptions.StaleDevicesException;
+import org.whispersystems.textsecure.internal.util.Base64;
 import org.whispersystems.textsecure.internal.util.BlacklistingTrustManager;
+import org.whispersystems.textsecure.internal.util.Util;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -54,10 +55,7 @@ import java.io.UnsupportedEncodingException;
 import java.net.HttpURLConnection;
 import java.net.URL;
 import java.security.KeyManagementException;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
-import java.security.cert.CertificateException;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
@@ -65,7 +63,6 @@ import java.util.Set;
 import javax.net.ssl.HttpsURLConnection;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManager;
-import javax.net.ssl.TrustManagerFactory;
 
 /**
  *
@@ -96,23 +93,20 @@ public class PushServiceSocket {
 
   private static final boolean ENFORCE_SSL = true;
 
-  private final String         serviceUrl;
-  private final String         localNumber;
-  private final String         password;
-  private final TrustManager[] trustManagers;
+  private final String              serviceUrl;
+  private final TrustManager[]      trustManagers;
+  private final CredentialsProvider credentialsProvider;
 
-  public PushServiceSocket(String serviceUrl, TrustStore trustStore,
-                           String localNumber, String password)
+  public PushServiceSocket(String serviceUrl, TrustStore trustStore, CredentialsProvider credentialsProvider)
   {
-    this.serviceUrl    = serviceUrl;
-    this.localNumber   = localNumber;
-    this.password      = password;
-    this.trustManagers = initializeTrustManager(trustStore);
+    this.serviceUrl          = serviceUrl;
+    this.credentialsProvider = credentialsProvider;
+    this.trustManagers       = BlacklistingTrustManager.createFor(trustStore);
   }
 
   public void createAccount(boolean voice) throws IOException {
     String path = voice ? CREATE_ACCOUNT_VOICE_PATH : CREATE_ACCOUNT_SMS_PATH;
-    makeRequest(String.format(path, localNumber), "GET", null);
+    makeRequest(String.format(path, credentialsProvider.getUser()), "GET", null);
   }
 
   public void verifyAccount(String verificationCode, String signalingKey,
@@ -145,7 +139,7 @@ public class PushServiceSocket {
   }
 
   public void registerGcmId(String gcmRegistrationId) throws IOException {
-    GcmRegistrationId registration = new GcmRegistrationId(gcmRegistrationId);
+    GcmRegistrationId registration = new GcmRegistrationId(gcmRegistrationId, true);
     makeRequest(REGISTER_GCM_PATH, "PUT", new Gson().toJson(registration));
   }
 
@@ -510,7 +504,7 @@ public class PushServiceSocket {
       connection.setRequestMethod(method);
       connection.setRequestProperty("Content-Type", "application/json");
 
-      if (password != null) {
+      if (credentialsProvider.getPassword() != null) {
         connection.setRequestProperty("Authorization", getAuthorizationHeader());
       }
 
@@ -539,35 +533,21 @@ public class PushServiceSocket {
 
   private String getAuthorizationHeader() {
     try {
-      return "Basic " + Base64.encodeBytes((localNumber + ":" + password).getBytes("UTF-8"));
+      return "Basic " + Base64.encodeBytes((credentialsProvider.getUser() + ":" + credentialsProvider.getPassword()).getBytes("UTF-8"));
     } catch (UnsupportedEncodingException e) {
       throw new AssertionError(e);
     }
   }
 
-  private TrustManager[] initializeTrustManager(TrustStore trustStore) {
-    try {
-      InputStream keyStoreInputStream = trustStore.getKeyStoreInputStream();
-      KeyStore    keyStore            = KeyStore.getInstance("BKS");
-
-      keyStore.load(keyStoreInputStream, trustStore.getKeyStorePassword().toCharArray());
-
-      TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
-      trustManagerFactory.init(keyStore);
-
-      return BlacklistingTrustManager.createFor(trustManagerFactory.getTrustManagers());
-    } catch (KeyStoreException | CertificateException | NoSuchAlgorithmException | IOException kse) {
-      throw new AssertionError(kse);
-    }
-  }
-
   private static class GcmRegistrationId {
     private String gcmRegistrationId;
+    private boolean webSocketChannel;
 
     public GcmRegistrationId() {}
 
-    public GcmRegistrationId(String gcmRegistrationId) {
+    public GcmRegistrationId(String gcmRegistrationId, boolean webSocketChannel) {
       this.gcmRegistrationId = gcmRegistrationId;
+      this.webSocketChannel  = webSocketChannel;
     }
   }
 

libtextsecure/src/main/java/org/whispersystems/textsecure/internal/util/BlacklistingTrustManager.java

@@ -16,13 +16,21 @@
  */
 package org.whispersystems.textsecure.internal.util;
 
+import org.whispersystems.textsecure.api.push.TrustStore;
+
+import java.io.IOException;
+import java.io.InputStream;
 import java.math.BigInteger;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
 import java.security.cert.CertificateException;
 import java.security.cert.X509Certificate;
 import java.util.LinkedList;
 import java.util.List;
 
 import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
 import javax.net.ssl.X509TrustManager;
 
 /**
@@ -51,6 +59,22 @@ public class BlacklistingTrustManager implements X509TrustManager {
     throw new AssertionError("No X509 Trust Managers!");
   }
 
+  public static TrustManager[] createFor(TrustStore trustStore) {
+    try {
+      InputStream keyStoreInputStream = trustStore.getKeyStoreInputStream();
+      KeyStore keyStore            = KeyStore.getInstance("BKS");
+
+      keyStore.load(keyStoreInputStream, trustStore.getKeyStorePassword().toCharArray());
+
+      TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+      trustManagerFactory.init(keyStore);
+
+      return BlacklistingTrustManager.createFor(trustManagerFactory.getTrustManagers());
+    } catch (KeyStoreException | CertificateException | IOException | NoSuchAlgorithmException e) {
+      throw new AssertionError(e);
+    }
+  }
+
   private final X509TrustManager trustManager;
 
   public BlacklistingTrustManager(X509TrustManager trustManager) {

libtextsecure/src/main/java/org/whispersystems/textsecure/internal/util/StaticCredentialsProvider.java

@@ -0,0 +1,31 @@
+package org.whispersystems.textsecure.internal.util;
+
+import org.whispersystems.textsecure.api.util.CredentialsProvider;
+
+public class StaticCredentialsProvider implements CredentialsProvider {
+
+  private final String user;
+  private final String password;
+  private final String signalingKey;
+
+  public StaticCredentialsProvider(String user, String password, String signalingKey) {
+    this.user         = user;
+    this.password     = password;
+    this.signalingKey = signalingKey;
+  }
+
+  @Override
+  public String getUser() {
+    return user;
+  }
+
+  @Override
+  public String getPassword() {
+    return password;
+  }
+
+  @Override
+  public String getSignalingKey() {
+    return signalingKey;
+  }
+}

libtextsecure/src/main/java/org/whispersystems/textsecure/internal/util/Util.java

@@ -102,4 +102,20 @@ public class Util {
     out.close();
   }
 
+  public static void sleep(long millis) {
+    try {
+      Thread.sleep(millis);
+    } catch (InterruptedException e) {
+      throw new AssertionError(e);
+    }
+  }
+
+  public static void wait(Object lock, long millis) {
+    try {
+      lock.wait(millis);
+    } catch (InterruptedException e) {
+      throw new AssertionError(e);
+    }
+  }
+
 }

libtextsecure/src/main/java/org/whispersystems/textsecure/internal/websocket/WebSocketConnection.java

@@ -0,0 +1,292 @@
+package org.whispersystems.textsecure.internal.websocket;
+
+import android.util.Log;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.squareup.okhttp.OkHttpClient;
+import com.squareup.okhttp.Request;
+import com.squareup.okhttp.Response;
+import com.squareup.okhttp.internal.ws.WebSocket;
+import com.squareup.okhttp.internal.ws.WebSocketListener;
+
+import org.whispersystems.textsecure.api.push.TrustStore;
+import org.whispersystems.textsecure.api.util.CredentialsProvider;
+import org.whispersystems.textsecure.internal.util.BlacklistingTrustManager;
+import org.whispersystems.textsecure.internal.util.Util;
+
+import java.io.IOException;
+import java.security.KeyManagementException;
+import java.security.NoSuchAlgorithmException;
+import java.util.LinkedList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+
+import okio.Buffer;
+import okio.BufferedSource;
+
+import static org.whispersystems.textsecure.internal.websocket.WebSocketProtos.WebSocketMessage;
+import static org.whispersystems.textsecure.internal.websocket.WebSocketProtos.WebSocketRequestMessage;
+import static org.whispersystems.textsecure.internal.websocket.WebSocketProtos.WebSocketResponseMessage;
+
+public class WebSocketConnection {
+
+  private static final String TAG = WebSocketConnection.class.getSimpleName();
+
+  private final LinkedList<WebSocketRequestMessage> incomingRequests = new LinkedList<>();
+
+  private final String              wsUri;
+  private final TrustStore          trustStore;
+  private final CredentialsProvider credentialsProvider;
+
+  private Client          client;
+  private KeepAliveSender keepAliveSender;
+
+  public WebSocketConnection(String httpUri, TrustStore trustStore, CredentialsProvider credentialsProvider) {
+    this.trustStore          = trustStore;
+    this.credentialsProvider = credentialsProvider;
+    this.wsUri               = httpUri.replace("https://", "wss://")
+                                      .replace("http://", "ws://") + "/v1/websocket/?login=%s&password=%s";
+  }
+
+  public synchronized void connect() {
+    Log.w(TAG, "WSC connect()...");
+
+    if (client == null) {
+      client = new Client(wsUri, trustStore, credentialsProvider);
+      client.connect();
+    }
+  }
+
+  public synchronized void disconnect() throws IOException {
+    Log.w(TAG, "WSC disconnect()...");
+
+    if (client != null) {
+      client.disconnect();
+      client = null;
+    }
+
+    if (keepAliveSender != null) {
+      keepAliveSender.shutdown();
+      keepAliveSender = null;
+    }
+  }
+
+  public synchronized WebSocketRequestMessage readRequest(long timeoutMillis)
+      throws TimeoutException, IOException
+  {
+    if (client == null) {
+      throw new IOException("Connection closed!");
+    }
+
+    long startTime = System.currentTimeMillis();
+
+    while (client != null && incomingRequests.isEmpty() && elapsedTime(startTime) < timeoutMillis) {
+      Util.wait(this, Math.max(1, timeoutMillis - elapsedTime(startTime)));
+    }
+
+    if      (incomingRequests.isEmpty() && client == null) throw new IOException("Connection closed!");
+    else if (incomingRequests.isEmpty())                   throw new TimeoutException("Timeout exceeded");
+    else                                                   return incomingRequests.removeFirst();
+  }
+
+  public synchronized void sendResponse(WebSocketResponseMessage response) throws IOException {
+    if (client == null) {
+      throw new IOException("Connection closed!");
+    }
+
+    WebSocketMessage message = WebSocketMessage.newBuilder()
+                                               .setType(WebSocketMessage.Type.RESPONSE)
+                                               .setResponse(response)
+                                               .build();
+
+    client.sendMessage(message.toByteArray());
+  }
+
+  private synchronized void sendKeepAlive() throws IOException {
+    if (keepAliveSender != null) {
+      client.sendMessage(WebSocketMessage.newBuilder()
+                                         .setType(WebSocketMessage.Type.REQUEST)
+                                         .setRequest(WebSocketRequestMessage.newBuilder()
+                                                                            .setId(System.currentTimeMillis())
+                                                                            .setPath("/v1/keepalive")
+                                                                            .setVerb("GET")
+                                                                            .build()).build()
+                                         .toByteArray());
+    }
+  }
+
+  private synchronized void onMessage(byte[] payload) {
+    Log.w(TAG, "WSC onMessage()");
+    try {
+      WebSocketMessage message = WebSocketMessage.parseFrom(payload);
+
+      Log.w(TAG, "Message Type: " + message.getType().getNumber());
+
+      if (message.getType().getNumber() == WebSocketMessage.Type.REQUEST_VALUE)  {
+        incomingRequests.add(message.getRequest());
+      }
+
+      notifyAll();
+    } catch (InvalidProtocolBufferException e) {
+      Log.w(TAG, e);
+    }
+  }
+
+  private synchronized void onClose() {
+    Log.w(TAG, "onClose()...");
+
+    if (client != null) {
+      client = null;
+      connect();
+    }
+
+    if (keepAliveSender != null) {
+      keepAliveSender.shutdown();
+      keepAliveSender = null;
+    }
+
+    notifyAll();
+  }
+
+  private synchronized void onConnected() {
+    if (client != null) {
+      keepAliveSender = new KeepAliveSender();
+      keepAliveSender.start();
+    }
+  }
+
+  private long elapsedTime(long startTime) {
+    return System.currentTimeMillis() - startTime;
+  }
+
+  private class Client implements WebSocketListener {
+
+    private final String              uri;
+    private final TrustStore          trustStore;
+    private final CredentialsProvider credentialsProvider;
+
+    private WebSocket webSocket;
+    private boolean   closed;
+
+    public Client(String uri, TrustStore trustStore, CredentialsProvider credentialsProvider) {
+      Log.w(TAG, "Connecting to: " + uri);
+
+      this.uri                 = uri;
+      this.trustStore          = trustStore;
+      this.credentialsProvider = credentialsProvider;
+    }
+
+    public void connect() {
+      new Thread() {
+        @Override
+        public void run() {
+          int attempt = 0;
+
+          while (newSocket()) {
+            try {
+              Response response = webSocket.connect(Client.this);
+
+              if (response.code() == 101) {
+                onConnected();
+                return;
+              }
+
+              Log.w(TAG, "WebSocket Response: " + response.code());
+            } catch (IOException e) {
+              Log.w(TAG, e);
+            }
+
+            Util.sleep(Math.min(++attempt * 200, TimeUnit.SECONDS.toMillis(15)));
+          }
+        }
+      }.start();
+    }
+
+    public synchronized void disconnect() {
+      Log.w(TAG, "Calling disconnect()...");
+      try {
+        closed = true;
+        if (webSocket != null) {
+          webSocket.close(1000, "OK");
+        }
+      } catch (IOException e) {
+        Log.w(TAG, e);
+      }
+    }
+
+    public void sendMessage(byte[] message) throws IOException {
+      webSocket.sendMessage(WebSocket.PayloadType.BINARY, new Buffer().write(message));
+    }
+
+    @Override
+    public void onMessage(BufferedSource payload, WebSocket.PayloadType type) throws IOException {
+      Log.w(TAG, "onMessage: " + type);
+      if (type.equals(WebSocket.PayloadType.BINARY)) {
+        WebSocketConnection.this.onMessage(payload.readByteArray());
+      }
+
+      payload.close();
+    }
+
+    @Override
+    public void onClose(int code, String reason) {
+      Log.w(TAG, String.format("onClose(%d, %s)", code, reason));
+      WebSocketConnection.this.onClose();
+    }
+
+    @Override
+    public void onFailure(IOException e) {
+      Log.w(TAG, e);
+      WebSocketConnection.this.onClose();
+    }
+
+    private synchronized boolean newSocket() {
+      if (closed) return false;
+
+      String           filledUri     = String.format(uri, credentialsProvider.getUser(), credentialsProvider.getPassword());
+      SSLSocketFactory socketFactory = createTlsSocketFactory(trustStore);
+
+      this.webSocket = WebSocket.newWebSocket(new OkHttpClient().setSslSocketFactory(socketFactory),
+                                              new Request.Builder().url(filledUri).build());
+
+      return true;
+    }
+
+    private SSLSocketFactory createTlsSocketFactory(TrustStore trustStore) {
+      try {
+        SSLContext context = SSLContext.getInstance("TLS");
+        context.init(null, BlacklistingTrustManager.createFor(trustStore), null);
+
+        return context.getSocketFactory();
+      } catch (NoSuchAlgorithmException | KeyManagementException e) {
+        throw new AssertionError(e);
+      }
+    }
+  }
+
+  private class KeepAliveSender extends Thread {
+
+    private AtomicBoolean stop = new AtomicBoolean(false);
+
+    public void run() {
+      while (!stop.get()) {
+        try {
+          Thread.sleep(TimeUnit.SECONDS.toMillis(15));
+
+          Log.w(TAG, "Sending keep alive...");
+          sendKeepAlive();
+        } catch (Throwable e) {
+          Log.w(TAG, e);
+        }
+      }
+    }
+
+    public void shutdown() {
+      stop.set(true);
+    }
+  }
+}