Working on StompClient. This version is not stable

forresthopkinsa · forresthopkinsa · commit a5c7586a9824 · 2017-08-04T09:16:14.000-07:00
diff --git a/README.md b/README.md
@@ -134,18 +134,89 @@ client.lifecycle().subscribe(lifecycleEvent -> {
 
 **Custom client**
 
-You can use a custom HttpClient (for example, if you want to allow untrusted HTTPS) using the four-argument overload of Stomp.over, like so:
+You can use a custom OkHttpClient (for example, [if you want to allow untrusted HTTPS](getUnsafeOkHttpClient())) using the four-argument overload of Stomp.over, like so:
 
 ``` java
-client = Stomp.over(WebSocket.class, address, null, unsafeClient);
+client = Stomp.over(WebSocket.class, address, null, getUnsafeOkHttpClient());
 ```
 
 Yes, it's safe to pass `null` for either (or both) of the last two arguments. That's exactly what the shorter overloads do.
 
+Note: This method is only supported using OkHttp, not JWS.
+
 **Support**
 
 Right now, the library only supports sending and receiving messages. ACK messages and transactions are not implemented yet.
 
+## Changes in this fork
+
+**Build changes**
+
+The upstream master is based on Retrolambda. This version is based on Native Java 8 compilation,
+although it should also work with Jack. It will *not* work with Retrolambda.
+
+**Code changes**
+
+These are the possible changes you need to make to your code for this branch, if you were using the upstream before:
+
+- Disconnecting is now mandatory
+  - Previously, the socket would automatically disconnect if nothing was listening to it
+  - Now, it will not disconnect unless you explicitly run client.disconnect()
+- send() now returns a Completable
+  - Previously, it returned an Observable\<Void\>
+  - Now, it and all its overloads return a Completable, which is functionally about the same
+  - **However**, Completable does not inherit from Observable, so there are a couple differences:
+    - Of course, if you were storing send() in an Observable, you'll have to change that
+    - Additionally, if you were inheriting onNext before, you're going to have to adjust it:
+      - Old way, deprecated:
+        ``` java
+        client.send("/app/hello", "world").subscribe(
+            aVoid -> Log.d(TAG, "Sent data!"),
+            error -> Log.e(TAG, "Encountered error while sending data!", error)
+        );
+        ```
+        !! -v:
+        ``` java
+        client.send("/app/hello", "world").subscribe(new Subscriber<Void>() {
+            @Override
+            public void onNext(Void aVoid) {
+                Log.d(TAG, "Sent data!");
+            }
+            @Override
+            public void onError(Throwable error) {
+                Log.e(TAG, "Encountered error while sending data!", error);
+            }
+            @Override
+            public void onCompleted() {} // useless
+        });
+        ```
+      - New way of handling it:
+        ``` java
+        client.send("/app/hello", "world").subscribe(
+            () -> Log.d(TAG, "Sent data!"),
+            error -> Log.e(TAG, "Encountered error while sending data!", error)
+        );
+        ```
+        Or if you just can't get enough of anonymous classes:
+        ``` java
+        client.send("/app/hello", "world").subscribe(new Subscriber<Object /*This can be anything*/ >() {
+            @Override
+            public void onCompleted() {
+                Log.d(TAG, "Sent data!");
+            }
+            @Override
+            public void onError(Throwable error) {
+                Log.e(TAG, "Encountered error while sending data!", error);
+            }
+            @Override
+            public void onNext(Object o) {} // useless
+        });
+        ```
+    - Be sure to implement this change, because the IDE might not catch the error.
+- 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
+
 ## Additional Reading
 
 - [Spring + Websockets + STOMP](https://spring.io/guides/gs/messaging-stomp-websocket/)
diff --git a/lib/src/main/java/ua/naiksoftware/stomp/client/StompClient.java b/lib/src/main/java/ua/naiksoftware/stomp/client/StompClient.java
@@ -13,10 +13,12 @@
 import java.util.UUID;
 import java.util.concurrent.CopyOnWriteArrayList;
 
+import rx.Completable;
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
 import rx.observables.ConnectableObservable;
+import rx.subjects.PublishSubject;
 import ua.naiksoftware.stomp.ConnectionProvider;
 import ua.naiksoftware.stomp.LifecycleEvent;
 import ua.naiksoftware.stomp.StompHeader;
@@ -31,17 +33,22 @@ public class StompClient {
     public static final String SUPPORTED_VERSIONS = "1.1,1.0";
     public static final String DEFAULT_ACK = "auto";
 
+    /*
     private Subscription mMessagesSubscription;
     private Map<String, Set<Subscriber<? super StompMessage>>> mSubscribers = new HashMap<>();
-    private List<ConnectableObservable<Void>> mWaitConnectionObservables;
+    */
+    private List<Completable> mWaitConnectionCompletables;
     private final ConnectionProvider mConnectionProvider;
     private HashMap<String, String> mTopics;
     private boolean mConnected;
     private boolean isConnecting;
 
+    private PublishSubject<StompMessage> mMessageStream;
+
     public StompClient(ConnectionProvider connectionProvider) {
         mConnectionProvider = connectionProvider;
-        mWaitConnectionObservables = new CopyOnWriteArrayList<>();
+        mWaitConnectionCompletables = new CopyOnWriteArrayList<>();
+        mMessageStream = PublishSubject.create();
     }
 
     /**
@@ -96,46 +103,43 @@ public void connect(List<StompHeader> _headers, boolean reconnect) {
                 });
 
         isConnecting = true;
-        mMessagesSubscription = mConnectionProvider.messages()
+        mConnectionProvider.messages()
                 .map(StompMessage::from)
+                .doOnNext(this::callSubscribers)
+                .filter(msg -> msg.getStompCommand().equals(StompCommand.CONNECTED))
                 .subscribe(stompMessage -> {
-                    if (stompMessage.getStompCommand().equals(StompCommand.CONNECTED)) {
-                        mConnected = true;
-                        isConnecting = false;
-                        for (ConnectableObservable<Void> observable : mWaitConnectionObservables) {
-                            observable.connect();
-                        }
-                        mWaitConnectionObservables.clear();
+                    mConnected = true;
+                    isConnecting = false;
+                    for (Completable completable : mWaitConnectionCompletables) {
+                        completable.subscribe();
                     }
-                    callSubscribers(stompMessage);
+                    mWaitConnectionCompletables.clear();
                 });
     }
 
-    public Observable<Void> send(String destination) {
+    public Completable send(String destination) {
         return send(new StompMessage(
                 StompCommand.SEND,
                 Collections.singletonList(new StompHeader(StompHeader.DESTINATION, destination)),
                 null));
     }
 
-    public Observable<Void> send(String destination, String data) {
+    public Completable send(String destination, String data) {
         return send(new StompMessage(
                 StompCommand.SEND,
                 Collections.singletonList(new StompHeader(StompHeader.DESTINATION, destination)),
                 data));
     }
 
-    public Observable<Void> send(StompMessage stompMessage) {
-        Observable<Void> observable = mConnectionProvider.send(stompMessage.compile()).toObservable();
+    public Completable send(StompMessage stompMessage) {
+        Completable completable = mConnectionProvider.send(stompMessage.compile());
         if (!mConnected) {
-            ConnectableObservable<Void> deferred = observable.publish();
-            mWaitConnectionObservables.add(deferred);
-            return deferred;
-        } else {
-            return observable;
+            mWaitConnectionCompletables.add(completable);
         }
+        return completable;
     }
 
+    /*
     private void callSubscribers(StompMessage stompMessage) {
         String messageDestination = stompMessage.findHeader(StompHeader.DESTINATION);
         for (String dest : mSubscribers.keySet()) {
@@ -147,6 +151,11 @@ private void callSubscribers(StompMessage stompMessage) {
             }
         }
     }
+    */
+
+    private void callSubscribers(StompMessage stompMessage) {
+        mMessageStream.onNext(stompMessage);
+    }
 
     public Observable<LifecycleEvent> lifecycle() {
         return mConnectionProvider.getLifecycleReceiver();
@@ -160,53 +169,95 @@ public Observable<StompMessage> topic(String destinationPath) {
         return topic(destinationPath, null);
     }
 
+    public Observable<StompMessage> topic(String destPath, List<StompHeader> headerList) {
+        Observable<StompMessage> ret;
+
+        if (destPath == null)
+            ret = Observable.error(new IllegalArgumentException("Topic path cannot be null"));
+        else
+            ret = mMessageStream
+                .filter(msg -> destPath.equals(msg.findHeader(StompHeader.DESTINATION)))
+                .doOnSubscribe(() -> subscribePath(destPath, headerList));
+        // still need to figure out how to do the unsubscribes reactively... more difficult than it sounds
+        return ret;
+    }
+
+    /*
     public Observable<StompMessage> topic(String destinationPath, List<StompHeader> headerList) {
+        // basically:
+        // on SUBSCRIBE, add the observer to the Set in the mSubscribers map that's associated with the specified topic,
+        // and send a subscribe message IF WE HAVEN'T ALREADY SUBSCRIBED TO THE TOPIC
+        //
+        // on UNSUBSCRIBE, remove unsubscribed observers, and remove unobserved topics
+
+        // on observer subscribe...
        return Observable.<StompMessage>create(subscriber -> {
+           // get list of other subscribers to topic
            Set<Subscriber<? super StompMessage>> subscribersSet = mSubscribers.get(destinationPath);
+           // if there are no other subscribers on topic...
            if (subscribersSet == null) {
+               // create new subscriber list,
                subscribersSet = new HashSet<>();
+               // and add the list to the map
                mSubscribers.put(destinationPath, subscribersSet);
+               // send SUBSCRIBE message and add topic to mTopics
                subscribePath(destinationPath, headerList).subscribe();
            }
+           // finally, now that we know that there is a list for this topic, add observer to it
            subscribersSet.add(subscriber);
 
        }).doOnUnsubscribe(() -> {
+           // on unsubscribe...
            Iterator<String> mapIterator = mSubscribers.keySet().iterator();
+           // for each topic in the map,
            while (mapIterator.hasNext()) {
+               // get topic path
                String destinationUrl = mapIterator.next();
+               // get observers subscribed to this topic
                Set<Subscriber<? super StompMessage>> set = mSubscribers.get(destinationUrl);
                Iterator<Subscriber<? super StompMessage>> setIterator = set.iterator();
+               // for each observer subscribed to this topic,
                while (setIterator.hasNext()) {
                    Subscriber<? super StompMessage> subscriber = setIterator.next();
+                   // if observer is no longer subscribed,
                    if (subscriber.isUnsubscribed()) {
+                       // remove it from the set
                        setIterator.remove();
+                       // if there are no observers subscribed to this topic anymore...
                        if (set.size() < 1) {
+                           // remote the set from the map
                            mapIterator.remove();
+                           // send UNSUBSCRIBE message
                            unsubscribePath(destinationUrl).subscribe();
                        }
                    }
                }
            }
        });
    }
+   */
+
+    private Completable subscribePath(String destinationPath, List<StompHeader> headerList) {
+        String topicId = UUID.randomUUID().toString();
 
-    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<>();
 
-          if (mTopics == null) mTopics = new HashMap<>();
-          mTopics.put(destinationPath, topicId);
-          List<StompHeader> headers = new ArrayList<>();
-          headers.add(new StompHeader(StompHeader.ID, topicId));
-          headers.add(new StompHeader(StompHeader.DESTINATION, destinationPath));
-          headers.add(new StompHeader(StompHeader.ACK, DEFAULT_ACK));
-          if (headerList != null) headers.addAll(headerList);
-          return send(new StompMessage(StompCommand.SUBSCRIBE,
-                  headers, null));
-      }
+        // Only continue if we don't already have a subscription to the topic
+        if (mTopics.containsKey(destinationPath))
+            return Completable.complete();
+
+        mTopics.put(destinationPath, topicId);
+        List<StompHeader> headers = new ArrayList<>();
+        headers.add(new StompHeader(StompHeader.ID, topicId));
+        headers.add(new StompHeader(StompHeader.DESTINATION, destinationPath));
+        headers.add(new StompHeader(StompHeader.ACK, DEFAULT_ACK));
+        if (headerList != null) headers.addAll(headerList);
+        return send(new StompMessage(StompCommand.SUBSCRIBE,
+                headers, null));
+    }
 
 
-    private Observable<Void> unsubscribePath(String dest) {
+    private Completable unsubscribePath(String dest) {
         String topicId = mTopics.get(dest);
         Log.d(TAG, "Unsubscribe path: " + dest + " id: " + topicId);
 
