Abstracted reconnect logic and implemented it for JWS

(JWS is still totally untested)

Author: forresthopkinsa

lib/src/main/java/ua/naiksoftware/stomp/AbstractConnectionProvider.java

@@ -4,8 +4,11 @@
 import android.support.annotation.Nullable;
 import android.util.Log;
 
+import java.util.concurrent.TimeUnit;
+
 import rx.Completable;
 import rx.Observable;
+import rx.subjects.BehaviorSubject;
 import rx.subjects.PublishSubject;
 
 /**
@@ -22,29 +25,56 @@ abstract class AbstractConnectionProvider implements ConnectionProvider {
     private final PublishSubject<LifecycleEvent> mLifecycleStream;
     @NonNull
     private final PublishSubject<String> mMessagesStream;
+    final BehaviorSubject<Boolean> mConnectionStream;
 
     AbstractConnectionProvider() {
         mLifecycleStream = PublishSubject.create();
         mMessagesStream = PublishSubject.create();
+        mConnectionStream = BehaviorSubject.create(false);
     }
 
     @NonNull
     @Override
     public Observable<String> messages() {
-        createWebSocketConnection();
-        return mMessagesStream;
+        return mMessagesStream.startWith(initSocket().toObservable());
     }
 
     /**
-     * Completable to close socket.
+     * Simply close socket.
      * <p>
      * For example:
      * <pre>
-     * return Completable.fromAction(() -> webSocket.close());
+     * webSocket.close();
      * </pre>
      */
+    abstract void rawDisconnect();
+
     @Override
-    public abstract Completable disconnect();
+    public Completable disconnect() {
+        Observable<Boolean> ex = Observable.error(new IllegalStateException("Attempted to disconnect when already disconnected"));
+
+        Completable block = mConnectionStream
+                .first(isConnected -> isConnected)
+                .timeout(1, TimeUnit.SECONDS, ex)
+                .toCompletable();
+
+        return Completable
+                .fromAction(this::rawDisconnect)
+                .startWith(block);
+    }
+
+    private Completable initSocket() {
+        Observable<Boolean> ex = Observable.error(new IllegalStateException("Attempted to connect when already connected"));
+
+        Completable block = mConnectionStream
+                .first(isConnected -> !isConnected)
+                .timeout(1, TimeUnit.SECONDS, ex)
+                .toCompletable();
+
+        return Completable
+                .fromAction(this::createWebSocketConnection)
+                .startWith(block);
+    }
 
     /**
      * Most important method: connects to websocket and notifies program of messages.
@@ -61,7 +91,7 @@ public Completable send(String stompMessage) {
                 throw new IllegalStateException("Not connected yet");
             } else {
                 Log.d(TAG, "Send STOMP message: " + stompMessage);
-                bareSend(stompMessage);
+                rawSend(stompMessage);
                 return null;
             }
         });
@@ -77,7 +107,7 @@ public Completable send(String stompMessage) {
      *
      * @param stompMessage message to send
      */
-    abstract void bareSend(String stompMessage);
+    abstract void rawSend(String stompMessage);
 
     /**
      * Get socket object.
@@ -94,6 +124,8 @@ public Completable send(String stompMessage) {
     void emitLifecycleEvent(@NonNull LifecycleEvent lifecycleEvent) {
         Log.d(TAG, "Emit lifecycle event: " + lifecycleEvent.getType().name());
         mLifecycleStream.onNext(lifecycleEvent);
+        if (lifecycleEvent.getType().equals(LifecycleEvent.Type.CLOSED))
+            mConnectionStream.onNext(false);
     }
 
     void emitMessage(String stompMessage) {

lib/src/main/java/ua/naiksoftware/stomp/OkHttpConnectionProvider.java

@@ -7,8 +7,6 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import okhttp3.Headers;
 import okhttp3.OkHttpClient;
@@ -17,8 +15,6 @@
 import okhttp3.WebSocket;
 import okhttp3.WebSocketListener;
 import okio.ByteString;
-import rx.Completable;
-import rx.subjects.BehaviorSubject;
 
 class OkHttpConnectionProvider extends AbstractConnectionProvider {
 
@@ -27,7 +23,6 @@ class OkHttpConnectionProvider extends AbstractConnectionProvider {
     private final Map<String, String> mConnectHttpHeaders;
     private final OkHttpClient mOkHttpClient;
     private final String tag = OkHttpConnectionProvider.class.getSimpleName();
-    private final BehaviorSubject<Boolean> mConnectionStream;
 
     @Nullable
     private WebSocket openedSocked;
@@ -37,51 +32,16 @@ class OkHttpConnectionProvider extends AbstractConnectionProvider {
         mUri = uri;
         mConnectHttpHeaders = connectHttpHeaders != null ? connectHttpHeaders : new HashMap<>();
         mOkHttpClient = okHttpClient;
-        mConnectionStream = BehaviorSubject.create(false);
-        mConnectionStream.subscribe(value -> Log.d(tag, "Connection stream emitted: " + value));
     }
 
     @NonNull
     @Override
-    public Completable disconnect() {
-        Completable block = mConnectionStream
-                .first(isConnected -> isConnected)
-                .timeout(1, TimeUnit.SECONDS)
-                .doOnError(error -> {
-                    if (error.getClass().equals(TimeoutException.class))
-                            Log.e(tag, "Attempted to disconnect when already disconnected");
-                })
-                .toCompletable();
-
-        /*
-        if (openedSocked == null) {
-            return Completable.error(new IllegalStateException("Attempted to disconnect when already disconnected"));
-        }
-        */
-
-        return Completable
-                .fromAction(() -> openedSocked.close(1000, ""))
-                .startWith(block);
+    public void rawDisconnect() {
+        openedSocked.close(1000, "");
     }
 
     @Override
     void createWebSocketConnection() {
-        Completable block = mConnectionStream
-                .first(isConnected -> !isConnected)
-                .timeout(1, TimeUnit.SECONDS)
-                .doOnError(error -> {
-                    if (error.getClass().equals(TimeoutException.class))
-                        Log.e(tag, "Attempted to connect when already connected");
-                })
-                .toCompletable();
-        block.get(); // todo: do this the right way
-
-        /*
-        if (openedSocked != null) {
-            throw new IllegalStateException("Already have connection to web socket");
-        }
-        */
-
         Request.Builder requestBuilder = new Request.Builder()
                 .url(mUri);
 
@@ -114,9 +74,8 @@ public void onMessage(WebSocket webSocket, @NonNull ByteString bytes) {
 
                     @Override
                     public void onClosed(WebSocket webSocket, int code, String reason) {
-                        emitLifecycleEvent(new LifecycleEvent(LifecycleEvent.Type.CLOSED));
                         openedSocked = null;
-                        mConnectionStream.onNext(false);
+                        emitLifecycleEvent(new LifecycleEvent(LifecycleEvent.Type.CLOSED));
                     }
 
                     @Override
@@ -130,7 +89,7 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) {
     }
 
     @Override
-    void bareSend(String stompMessage) {
+    void rawSend(String stompMessage) {
         openedSocked.send(stompMessage);
     }
 

lib/src/main/java/ua/naiksoftware/stomp/WebSocketsConnectionProvider.java

@@ -20,8 +20,6 @@
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLSocketFactory;
 
-import rx.Completable;
-
 /**
  * Created by naik on 05.05.16.
  */
@@ -49,8 +47,8 @@ class WebSocketsConnectionProvider extends AbstractConnectionProvider {
 
     @NonNull
     @Override
-    public Completable disconnect() {
-        return Completable.fromAction(() -> mWebSocketClient.close());
+    public void rawDisconnect() {
+        mWebSocketClient.close();
     }
 
     @Override
@@ -112,10 +110,11 @@ public void onError(Exception ex) {
 
         mWebSocketClient.connect();
         haveConnection = true;
+        mConnectionStream.onNext(true);
     }
 
     @Override
-    void bareSend(String stompMessage) {
+    void rawSend(String stompMessage) {
         mWebSocketClient.send(stompMessage);
     }