Working on reconnection

Author: forresthopkinsa

build.gradle

@@ -8,7 +8,7 @@ buildscript {
         }
     }
     dependencies {
-        classpath 'com.android.tools.build:gradle:3.0.0-alpha9'
+        classpath 'com.android.tools.build:gradle:3.0.0-beta1'
         classpath 'com.github.dcendents:android-maven-gradle-plugin:1.5'
 
         // NOTE: Do not place your application dependencies here; they belong

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

@@ -2,6 +2,7 @@
 
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
+import android.util.Log;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -22,6 +23,7 @@ class OkHttpConnectionProvider extends AbstractConnectionProvider {
     @NonNull
     private final Map<String, String> mConnectHttpHeaders;
     private final OkHttpClient mOkHttpClient;
+    private final String tag = OkHttpConnectionProvider.class.getSimpleName();
 
     @Nullable
     private WebSocket openedSocked;
@@ -36,7 +38,11 @@ class OkHttpConnectionProvider extends AbstractConnectionProvider {
     @NonNull
     @Override
     public Completable disconnect() {
-        return Completable.fromAction(() -> openedSocked.close(1000, ""));
+        if (openedSocked == null) {
+            return Completable.error(new IllegalStateException("Attempted to disconnect when already disconnected"));
+        }
+        return Completable
+                .fromAction(() -> openedSocked.close(1000, ""));
     }
 
     @Override
@@ -65,7 +71,10 @@ public void onOpen(WebSocket webSocket, @NonNull Response response) {
 
                     @Override
                     public void onMessage(WebSocket webSocket, String text) {
-                        emitMessage(text);
+                        if (text.equals("\n"))
+                            Log.d(tag, "RECEIVED HEARTBEAT");
+                        else
+                            emitMessage(text);
                     }
 
                     @Override

lib/src/main/java/ua/naiksoftware/stomp/client/StompClient.java

@@ -29,6 +29,7 @@ public class StompClient {
     public static final String SUPPORTED_VERSIONS = "1.1,1.0";
     public static final String DEFAULT_ACK = "auto";
 
+    private final String tag = StompClient.class.getSimpleName();
     private final ConnectionProvider mConnectionProvider;
     private HashMap<String, String> mTopics;
     private boolean mConnected;
@@ -42,9 +43,13 @@ public class StompClient {
     public StompClient(ConnectionProvider connectionProvider) {
         mConnectionProvider = connectionProvider;
         mMessageStream = PublishSubject.create();
+        mStreamMap = new HashMap<>();
+        resetStatus();
+    }
+
+    private void resetStatus() {
         mConnectionFuture = new CompletableFuture<>();
         mConnectionComplete = Completable.fromFuture(mConnectionFuture).subscribeOn(Schedulers.newThread());
-        mStreamMap = new HashMap<>();
     }
 
     /**
@@ -61,7 +66,7 @@ public void connect(boolean reconnect) {
     /**
      * Connect without reconnect if connected
      *
-     * @param _headers might be null
+     * @param _headers HTTP headers to send in the INITIAL REQUEST, i.e. during the protocol upgrade
      */
     public void connect(List<StompHeader> _headers) {
         connect(_headers, false);
@@ -70,7 +75,7 @@ public void connect(List<StompHeader> _headers) {
     /**
      * If already connected and reconnect=false - nope
      *
-     * @param _headers might be null
+     * @param _headers HTTP headers to send in the INITIAL REQUEST, i.e. during the protocol upgrade
      */
     public void connect(@Nullable List<StompHeader> _headers, boolean reconnect) {
         if (reconnect) disconnect();
@@ -123,7 +128,6 @@ public Completable send(String destination, String data) {
 
     public Completable send(@NonNull StompMessage stompMessage) {
         Completable completable = mConnectionProvider.send(stompMessage.compile());
-        mConnectionComplete.subscribe();
         return completable.startWith(mConnectionComplete);
     }
 
@@ -136,7 +140,8 @@ public Observable<LifecycleEvent> lifecycle() {
     }
 
     public void disconnect() {
-        mConnectionProvider.disconnect().subscribe();
+        resetStatus();
+        mConnectionProvider.disconnect().subscribe(() -> mConnected = false);
     }
 
     public Observable<StompMessage> topic(String destinationPath) {