Cleaning up, restoring JWS support

Author: forresthopkinsa

README.md

@@ -146,7 +146,7 @@ Yes, it's safe to pass `null` for either (or both) of the last two arguments. Th
 
 Right now, the library only supports sending and receiving messages. ACK messages and transactions are not implemented yet.
 
-**Additional Reading**
+## Additional Reading
 
 - [Spring + Websockets + STOMP](https://spring.io/guides/gs/messaging-stomp-websocket/)
 - [STOMP Protocol](http://stomp.github.io/)

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

@@ -18,7 +18,7 @@ public interface ConnectionProvider {
      * onError if not connected or error detected will be called, or onCompleted id sending started
      * TODO: send messages with ACK
      */
-    Observable<Void> send(String stompMessage);
+    Completable send(String stompMessage);
 
     /**
      * Subscribe this for receive #LifecycleEvent events

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

@@ -2,10 +2,7 @@
 
 import android.util.Log;
 
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -18,30 +15,24 @@
 import okio.ByteString;
 import rx.Completable;
 import rx.Observable;
-import rx.Subscriber;
 import rx.subjects.PublishSubject;
 
-/* package */ class OkHttpConnectionProvider implements ConnectionProvider {
+class OkHttpConnectionProvider implements ConnectionProvider {
 
     private static final String TAG = WebSocketsConnectionProvider.class.getSimpleName();
 
     private final String mUri;
     private final Map<String, String> mConnectHttpHeaders;
     private final OkHttpClient mOkHttpClient;
 
-    private final List<Subscriber<? super LifecycleEvent>> mLifecycleSubscribers;
     private final PublishSubject<LifecycleEvent> mLifecycleStream;
-    private final List<Subscriber<? super String>> mMessagesSubscribers;
     private final PublishSubject<String> mMessagesStream;
 
     private WebSocket openedSocked;
 
-
-    /* package */ OkHttpConnectionProvider(String uri, Map<String, String> connectHttpHeaders, OkHttpClient okHttpClient) {
+    OkHttpConnectionProvider(String uri, Map<String, String> connectHttpHeaders, OkHttpClient okHttpClient) {
         mUri = uri;
         mConnectHttpHeaders = connectHttpHeaders != null ? connectHttpHeaders : new HashMap<>();
-        mLifecycleSubscribers = new ArrayList<>();
-        mMessagesSubscribers = new ArrayList<>();
         mOkHttpClient = okHttpClient;
 
         mLifecycleStream = PublishSubject.create();
@@ -51,35 +42,9 @@
     @Override
     public Observable<String> messages() {
         createWebSocketConnection();
-        // By using Subjects, we can leave the tracking of Subscribers to Rx.
-        // Additionally, server disconnection is now handled manually
-        //    (instead of trying to support disconnecting just by unsubscribing)
         return mMessagesStream;
-
-        /*
-        Observable<String> observable = Observable.<String>create(subscriber -> {
-            mMessagesSubscribers.add(subscriber);
-
-        }).doOnUnsubscribe(() -> {
-            Iterator<Subscriber<? super String>> iterator = mMessagesSubscribers.iterator();
-            while (iterator.hasNext()) {
-                if (iterator.next().isUnsubscribed()) iterator.remove();
-            }
-
-            if (mMessagesSubscribers.size() < 1) {
-                Log.d(TAG, "Close web socket connection now in thread " + Thread.currentThread());
-                openedSocked.close(1000, "");
-                openedSocked = null;
-            }
-        });
-
-        createWebSocketConnection();
-        return observable;
-        */
     }
 
-    // this used to be done automatically whenever the "subscriber list" was empty
-    // this way is more discrete
     @Override
     public Completable disconnect() {
         return Completable.fromAction(() -> openedSocked.close(1000, ""));
@@ -134,9 +99,8 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) {
     }
 
     @Override
-    public Observable<Void> send(String stompMessage) {
-        // .create(onSubscribe) is deprecated because it's unsafe
-        return Observable.fromCallable(() -> {
+    public Completable send(String stompMessage) {
+        return Completable.fromCallable(() -> {
             if (openedSocked == null) {
                 throw new IllegalStateException("Not connected yet");
             } else {
@@ -149,20 +113,7 @@ public Observable<Void> send(String stompMessage) {
 
     @Override
     public Observable<LifecycleEvent> getLifecycleReceiver() {
-        // Once again, opting to leave Subscriber tracking to Rx
         return mLifecycleStream;
-
-        /*
-        return Observable.<LifecycleEvent>create(subscriber -> {
-            mLifecycleSubscribers.add(subscriber);
-
-        }).doOnUnsubscribe(() -> {
-            Iterator<Subscriber<? super LifecycleEvent>> iterator = mLifecycleSubscribers.iterator();
-            while (iterator.hasNext()) {
-                if (iterator.next().isUnsubscribed()) iterator.remove();
-            }
-        });
-        */
     }
 
     private TreeMap<String, String> headersAsMap(Response response) {
@@ -182,24 +133,11 @@ private void addConnectionHeadersToBuilder(Request.Builder requestBuilder, Map<S
 
     private void emitLifecycleEvent(LifecycleEvent lifecycleEvent) {
         Log.d(TAG, "Emit lifecycle event: " + lifecycleEvent.getType().name());
-        // I know Subjects are discouraged, but I think this is way cleaner than before
         mLifecycleStream.onNext(lifecycleEvent);
-
-        /*
-        for (Subscriber<? super LifecycleEvent> subscriber : mLifecycleSubscribers) {
-            subscriber.onNext(lifecycleEvent);
-        }
-        */
     }
 
     private void emitMessage(String stompMessage) {
         Log.d(TAG, "Emit STOMP message: " + stompMessage);
         mMessagesStream.onNext(stompMessage);
-
-        /*
-        for (Subscriber<? super String> subscriber : mMessagesSubscribers) {
-            subscriber.onNext(stompMessage);
-        }
-        */
     }
 }

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

@@ -10,10 +10,8 @@
 import org.java_websocket.handshake.ServerHandshake;
 
 import java.net.URI;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -22,55 +20,47 @@
 
 import rx.Completable;
 import rx.Observable;
-import rx.Subscriber;
+import rx.subjects.PublishSubject;
 
 /**
  * Created by naik on 05.05.16.
  */
-/* package */ class WebSocketsConnectionProvider implements ConnectionProvider {
+
+class WebSocketsConnectionProvider implements ConnectionProvider {
 
     private static final String TAG = WebSocketsConnectionProvider.class.getSimpleName();
 
     private final String mUri;
     private final Map<String, String> mConnectHttpHeaders;
 
-    private final List<Subscriber<? super LifecycleEvent>> mLifecycleSubscribers;
-    private final List<Subscriber<? super String>> mMessagesSubscribers;
-
     private WebSocketClient mWebSocketClient;
     private boolean haveConnection;
     private TreeMap<String, String> mServerHandshakeHeaders;
 
+    private final PublishSubject<LifecycleEvent> mLifecycleStream;
+    private final PublishSubject<String> mMessagesStream;
+
     /**
      * Support UIR scheme ws://host:port/path
      * @param connectHttpHeaders may be null
      */
-    /* package */ WebSocketsConnectionProvider(String uri, Map<String, String> connectHttpHeaders) {
+    WebSocketsConnectionProvider(String uri, Map<String, String> connectHttpHeaders) {
         mUri = uri;
         mConnectHttpHeaders = connectHttpHeaders != null ? connectHttpHeaders : new HashMap<>();
-        mLifecycleSubscribers = new ArrayList<>();
-        mMessagesSubscribers = new ArrayList<>();
+
+        mLifecycleStream = PublishSubject.create();
+        mMessagesStream = PublishSubject.create();
     }
 
     @Override
     public Observable<String> messages() {
-        Observable<String> observable = Observable.<String>create(subscriber -> {
-            mMessagesSubscribers.add(subscriber);
-
-        }).doOnUnsubscribe(() -> {
-            Iterator<Subscriber<? super String>> iterator = mMessagesSubscribers.iterator();
-            while (iterator.hasNext()) {
-                if (iterator.next().isUnsubscribed()) iterator.remove();
-            }
-
-            if (mMessagesSubscribers.size() < 1) {
-                Log.d(TAG, "Close web socket connection now in thread " + Thread.currentThread());
-                mWebSocketClient.close();
-            }
-        });
-
         createWebSocketConnection();
-        return observable;
+        return mMessagesStream;
+    }
+
+    @Override
+    public Completable disconnect() {
+        return Completable.fromAction(() -> mWebSocketClient.close());
     }
 
     private void createWebSocketConnection() {
@@ -134,50 +124,30 @@ public void onError(Exception ex) {
     }
 
     @Override
-    public Observable<Void> send(String stompMessage) {
-        return Observable.create(subscriber -> {
+    public Completable send(String stompMessage) {
+        return Completable.fromCallable(() -> {
             if (mWebSocketClient == null) {
-                subscriber.onError(new IllegalStateException("Not connected yet"));
+                throw new IllegalStateException("Not connected yet");
             } else {
                 Log.d(TAG, "Send STOMP message: " + stompMessage);
                 mWebSocketClient.send(stompMessage);
-                subscriber.onCompleted();
+                return null;
             }
         });
     }
 
-    // Just to appease javac
-    @Override
-    public Completable disconnect() {
-        return Completable.fromAction(() -> {
-            throw new UnsupportedOperationException("JAVA WEB SOCKETS ARE NOT YET SUPPORTED IN THIS VERSION");
-        });
-    }
-
     private void emitLifecycleEvent(LifecycleEvent lifecycleEvent) {
         Log.d(TAG, "Emit lifecycle event: " + lifecycleEvent.getType().name());
-        for (Subscriber<? super LifecycleEvent> subscriber : mLifecycleSubscribers) {
-            subscriber.onNext(lifecycleEvent);
-        }
+        mLifecycleStream.onNext(lifecycleEvent);
     }
 
     private void emitMessage(String stompMessage) {
         Log.d(TAG, "Emit STOMP message: " + stompMessage);
-        for (Subscriber<? super String> subscriber : mMessagesSubscribers) {
-            subscriber.onNext(stompMessage);
-        }
+        mMessagesStream.onNext(stompMessage);
     }
 
     @Override
     public Observable<LifecycleEvent> getLifecycleReceiver() {
-        return Observable.<LifecycleEvent>create(subscriber -> {
-            mLifecycleSubscribers.add(subscriber);
-
-        }).doOnUnsubscribe(() -> {
-            Iterator<Subscriber<? super LifecycleEvent>> iterator = mLifecycleSubscribers.iterator();
-            while (iterator.hasNext()) {
-                if (iterator.next().isUnsubscribed()) iterator.remove();
-            }
-        });
+        return mLifecycleStream;
     }
 }

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

@@ -72,7 +72,6 @@ public void connect(List<StompHeader> _headers) {
     public void connect(List<StompHeader> _headers, boolean reconnect) {
         if (reconnect) disconnect();
         if (mConnected) return;
-        // why wasn't this DRY before?
         lifecycle()
                 .subscribe(lifecycleEvent -> {
                     switch (lifecycleEvent.getType()) {
@@ -127,9 +126,8 @@ public Observable<Void> send(String destination, String data) {
     }
 
     public Observable<Void> send(StompMessage stompMessage) {
-        Observable<Void> observable = mConnectionProvider.send(stompMessage.compile());
+        Observable<Void> observable = mConnectionProvider.send(stompMessage.compile()).toObservable();
         if (!mConnected) {
-            // my inner grammar nazi
             ConnectableObservable<Void> deferred = observable.publish();
             mWaitConnectionObservables.add(deferred);
             return deferred;
@@ -155,7 +153,6 @@ public Observable<LifecycleEvent> lifecycle() {
     }
 
     public void disconnect() {
-        // the other things are now taken care of downstream
         mConnectionProvider.disconnect().subscribe();
     }