Successfully replaced the old mechanism for caching pre-connect sends.

There is a lot of debug logging in this build, but everything now seems to be working correctly.

Author: forresthopkinsa

lib/build.gradle

@@ -32,6 +32,8 @@ dependencies {
     compile fileTree(include: ['*.jar'], dir: 'libs')
     testCompile 'junit:junit:4.12'
     compile 'io.reactivex:rxjava:1.3.0'
+    compile 'net.sourceforge.streamsupport:streamsupport:1.5.5'
+    compile 'net.sourceforge.streamsupport:streamsupport-cfuture:1.5.5'
     // Supported transports
     compile 'org.java-websocket:java-websocket:1.3.2'
     compile 'com.squareup.okhttp3:okhttp:3.8.1'

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

@@ -19,7 +19,7 @@
 
 class OkHttpConnectionProvider implements ConnectionProvider {
 
-    private static final String TAG = WebSocketsConnectionProvider.class.getSimpleName();
+    private static final String TAG = OkHttpConnectionProvider.class.getSimpleName();
 
     private final String mUri;
     private final Map<String, String> mConnectHttpHeaders;

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

@@ -13,11 +13,13 @@
 import java.util.UUID;
 import java.util.concurrent.CopyOnWriteArrayList;
 
+import java8.util.concurrent.CompletableFuture;
 import rx.Completable;
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
 import rx.observables.ConnectableObservable;
+import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
 import ua.naiksoftware.stomp.ConnectionProvider;
 import ua.naiksoftware.stomp.LifecycleEvent;
@@ -37,18 +39,23 @@ public class StompClient {
     private Subscription mMessagesSubscription;
     private Map<String, Set<Subscriber<? super StompMessage>>> mSubscribers = new HashMap<>();
     */
-    private List<Completable> mWaitConnectionCompletables;
+//    private List<Completable> mWaitConnectionCompletables;
     private final ConnectionProvider mConnectionProvider;
     private HashMap<String, String> mTopics;
     private boolean mConnected;
     private boolean isConnecting;
 
     private PublishSubject<StompMessage> mMessageStream;
+    private CompletableFuture<Boolean> connectionStatus;
+    private Completable waitForConnect;
 
     public StompClient(ConnectionProvider connectionProvider) {
         mConnectionProvider = connectionProvider;
-        mWaitConnectionCompletables = new CopyOnWriteArrayList<>();
+//        mWaitConnectionCompletables = new CopyOnWriteArrayList<>();
         mMessageStream = PublishSubject.create();
+        connectionStatus = new CompletableFuture<>();
+        waitForConnect = Completable.fromFuture(connectionStatus).subscribeOn(Schedulers.newThread());
+        waitForConnect.subscribe(() -> Log.d(TAG, "waitForConnect completed"));
     }
 
     /**
@@ -87,7 +94,7 @@ public void connect(List<StompHeader> _headers, boolean reconnect) {
                             headers.add(new StompHeader(StompHeader.VERSION, SUPPORTED_VERSIONS));
                             if (_headers != null) headers.addAll(_headers);
                             mConnectionProvider.send(new StompMessage(StompCommand.CONNECT, headers, null).compile())
-                                    .subscribe();
+                                    .subscribe(() -> Log.d(TAG, "CONNECT command sent!"));
                             break;
 
                         case CLOSED:
@@ -110,18 +117,18 @@ public void connect(List<StompHeader> _headers, boolean reconnect) {
                 .subscribe(stompMessage -> {
                     mConnected = true;
                     isConnecting = false;
+                    connectionStatus.complete(true);
+                    /*
                     for (Completable completable : mWaitConnectionCompletables) {
                         completable.subscribe();
                     }
                     mWaitConnectionCompletables.clear();
+                    */
                 });
     }
 
     public Completable send(String destination) {
-        return send(new StompMessage(
-                StompCommand.SEND,
-                Collections.singletonList(new StompHeader(StompHeader.DESTINATION, destination)),
-                null));
+        return send(destination, null);
     }
 
     public Completable send(String destination, String data) {
@@ -133,10 +140,13 @@ public Completable send(String destination, String data) {
 
     public Completable send(StompMessage stompMessage) {
         Completable completable = mConnectionProvider.send(stompMessage.compile());
+        /*
         if (!mConnected) {
             mWaitConnectionCompletables.add(completable);
         }
-        return completable;
+        */
+        waitForConnect.subscribe(() -> Log.d(TAG, "SEND waitForConnect complete, continuing!"));
+        return completable.startWith(waitForConnect);
     }
 
     /*
@@ -177,7 +187,7 @@ public Observable<StompMessage> topic(String destPath, List<StompHeader> headerL
         else
             ret = mMessageStream
                 .filter(msg -> destPath.equals(msg.findHeader(StompHeader.DESTINATION)))
-                .doOnSubscribe(() -> subscribePath(destPath, headerList));
+                .doOnSubscribe(() -> subscribePath(destPath, headerList).subscribe());
         // still need to figure out how to do the unsubscribes reactively... more difficult than it sounds
         return ret;
     }