Update StompClient.java

Author: NaikSoftware

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

@@ -3,7 +3,6 @@
 import android.util.Log;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -122,10 +121,6 @@ public Observable<Void> send(StompMessage stompMessage) {
             mWaitConnectionObservables.add(deffered);
             return deffered;
         } else {
-            if(stompMessage.getStompCommand().equals(StompCommand.SUBSCRIBE)){
-                ConnectableObservable<Void> send = mConnectionProvider.send(stompMessage.compile()).publish();
-                send.connect();
-            }
             return observable;
         }
     }
@@ -161,7 +156,7 @@ public Observable<StompMessage> topic(String destinationPath, List<StompHeader>
            if (subscribersSet == null) {
                subscribersSet = new HashSet<>();
                mSubscribers.put(destinationPath, subscribersSet);
-               subscribePath(destinationPath, headerList);
+               subscribePath(destinationPath, headerList).subscribe();
            }
            subscribersSet.add(subscriber);
 
@@ -173,16 +168,16 @@ public Observable<StompMessage> topic(String destinationPath, List<StompHeader>
                        set.remove(subscriber);
                        if (set.size() < 1) {
                            mSubscribers.remove(dest);
-                           unsubscribePath(dest);
+                           unsubscribePath(dest).subscribe();
                        }
                    }
                }
            }
        });
    }
 
-    private void subscribePath(String destinationPath, List<StompHeader> headerList) {
-          if (destinationPath == null) return;
+    private Observable<Void> subscribePath(String destinationPath, List<StompHeader> headerList) {
+          if (destinationPath == null) return null;
           String topicId = UUID.randomUUID().toString();
 
           if (mTopics == null) mTopics = new HashMap<>();
@@ -192,16 +187,16 @@ private void subscribePath(String destinationPath, List<StompHeader> headerList)
           headers.add(new StompHeader(StompHeader.DESTINATION, destinationPath));
           headers.add(new StompHeader(StompHeader.ACK, DEFAULT_ACK));
           if (headerList != null) headers.addAll(headerList);
-          send(new StompMessage(StompCommand.SUBSCRIBE,
+          return send(new StompMessage(StompCommand.SUBSCRIBE,
                   headers, null));
       }
 
 
-    private void unsubscribePath(String dest) {
+    private Observable<Void> unsubscribePath(String dest) {
         String topicId = mTopics.get(dest);
         Log.d(TAG, "Unsubscribe path: " + dest + " id: " + topicId);
 
-        send(new StompMessage(StompCommand.UNSUBSCRIBE,
+        return send(new StompMessage(StompCommand.UNSUBSCRIBE,
                 Collections.singletonList(new StompHeader(StompHeader.ID, topicId)), null));
     }