Finished fixing reconnect and its race conditions

Managed to do it without any `synchronized` blocking

Author: forresthopkinsa

README.md

@@ -30,6 +30,8 @@ You can use this library two ways:
 - Using the new Native Java 8 support
   - As of this writing, you must be using Android Studio Canary to use this feature.
   - Has been tested in the following environments:
+    - Beta 2, Gradle plugin v3.0.0-beta2
+    - Beta 1, Gradle plugin v3.0.0-beta1
     - Canary 9, Gradle plugin v3.0.0-alpha9
     - Canary 8, Gradle plugin v3.0.0-alpha8
   - It *should* work in all 3.0.0+ versions
@@ -219,6 +221,21 @@ These are the possible changes you need to make to your code for this branch, if
         });
         ```
     - Be sure to implement this change, because the IDE might not catch the error.
+- Removed `reconnect` parameter from `connect(...)` methods
+  - Old, deprecated overloads of the method:
+    ``` java
+    void connect(boolean reconnect) {...}
+    void connect(List<StompHeader> _headers, boolean reconnect) {...}
+    ```
+  - Now, these are the *only* two ways to call `connect` (these existed before, too):
+    ``` java
+    void connect() {...}
+    void connect(List<StompHeader> _headers) {...}
+    ```
+  - Additionally, reconnection is now handled by just calling `reconnect()`
+    - It automatically attaches the last-used connect headers
+    - It is meant to be used when already connected; it executes `disconnect()`
+    - Note that if you're already disconnected, this will throw a `TimeoutException`
 - Passing null as the topic path now throws an exception
   - Previously, it was supposed to silently fail, although it would probably hit a NPE first (untested)
   - Now it throws an IllegalArgumentException

build.gradle

@@ -8,7 +8,7 @@ buildscript {
         }
     }
     dependencies {
-        classpath 'com.android.tools.build:gradle:3.0.0-beta1'
+        classpath 'com.android.tools.build:gradle:3.0.0-beta2'
         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

@@ -7,6 +7,8 @@
 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;
@@ -16,6 +18,7 @@
 import okhttp3.WebSocketListener;
 import okio.ByteString;
 import rx.Completable;
+import rx.subjects.BehaviorSubject;
 
 class OkHttpConnectionProvider extends AbstractConnectionProvider {
 
@@ -24,6 +27,7 @@ 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;
@@ -33,24 +37,50 @@ 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, ""));
+                .fromAction(() -> openedSocked.close(1000, ""))
+                .startWith(block);
     }
 
     @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);
@@ -86,6 +116,7 @@ public void onMessage(WebSocket webSocket, @NonNull ByteString bytes) {
                     public void onClosed(WebSocket webSocket, int code, String reason) {
                         emitLifecycleEvent(new LifecycleEvent(LifecycleEvent.Type.CLOSED));
                         openedSocked = null;
+                        mConnectionStream.onNext(false);
                     }
 
                     @Override
@@ -95,6 +126,7 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) {
                 }
 
         );
+        mConnectionStream.onNext(true);
     }
 
     @Override

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

@@ -14,6 +14,7 @@
 import java8.util.concurrent.CompletableFuture;
 import rx.Completable;
 import rx.Observable;
+import rx.Subscription;
 import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
 import ua.naiksoftware.stomp.ConnectionProvider;
@@ -41,6 +42,8 @@ public class StompClient {
     private Completable mConnectionComplete;
     private HashMap<String, Observable<StompMessage>> mStreamMap;
     private Parser parser;
+    private Subscription lifecycleSub;
+    private List<StompHeader> mHeaders;
 
     public StompClient(ConnectionProvider connectionProvider) {
         mConnectionProvider = connectionProvider;
@@ -82,28 +85,16 @@ public void connect() {
         connect(null);
     }
 
-    public void connect(boolean reconnect) {
-        connect(null, reconnect);
-    }
-
-    /**
-     * Connect without reconnect if connected
-     *
-     * @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);
-    }
-
     /**
      * If already connected and reconnect=false - nope
      *
      * @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();
+    public void connect(@Nullable List<StompHeader> _headers) {
+        mHeaders = _headers;
+
         if (mConnected) return;
-        lifecycle()
+        lifecycleSub = lifecycle()
                 .subscribe(lifecycleEvent -> {
                     switch (lifecycleEvent.getType()) {
                         case OPENED:
@@ -138,6 +129,14 @@ public void connect(@Nullable List<StompHeader> _headers, boolean reconnect) {
                 });
     }
 
+    /**
+     * Disconnect from server, and then reconnect with the last-used headers
+     */
+    public void reconnect() {
+        disconnect();
+        connect(mHeaders);
+    }
+
     public Completable send(String destination) {
         return send(destination, null);
     }
@@ -164,6 +163,7 @@ public Observable<LifecycleEvent> lifecycle() {
 
     public void disconnect() {
         resetStatus();
+        lifecycleSub.unsubscribe();
         mConnectionProvider.disconnect().subscribe(() -> mConnected = false);
     }