Merge pull request #32 from SarojMaharjan/master

Solution for issue #19

Author: NaikSoftware

.idea/markdown-navigator/profiles_settings.xml

@@ -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;
@@ -157,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);
 
@@ -169,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 Observable.empty();
           String topicId = UUID.randomUUID().toString();
 
           if (mTopics == null) mTopics = new HashMap<>();
@@ -188,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));
     }