Added disconnect() to interface, cleaned up some Rx methods

Author: forresthopkinsa

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

@@ -1,5 +1,6 @@
 package ua.naiksoftware.stomp;
 
+import rx.Completable;
 import rx.Observable;
 
 /**
@@ -23,4 +24,10 @@ public interface ConnectionProvider {
      * Subscribe this for receive #LifecycleEvent events
      */
     Observable<LifecycleEvent> getLifecycleReceiver();
+
+    /**
+     * Disconnects from server. This is basically a Callable.
+     * Automatically emits Lifecycle.CLOSE
+     */
+    Completable disconnect();
 }

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

@@ -16,8 +16,10 @@
 import okhttp3.WebSocket;
 import okhttp3.WebSocketListener;
 import okio.ByteString;
+import rx.Completable;
 import rx.Observable;
 import rx.Subscriber;
+import rx.subjects.PublishSubject;
 
 /* package */ class OkHttpConnectionProvider implements ConnectionProvider {
 
@@ -28,7 +30,9 @@
     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;
 
@@ -39,10 +43,20 @@
         mLifecycleSubscribers = new ArrayList<>();
         mMessagesSubscribers = new ArrayList<>();
         mOkHttpClient = okHttpClient;
+
+        mLifecycleStream = PublishSubject.create();
+        mMessagesStream = PublishSubject.create();
     }
 
     @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);
 
@@ -61,6 +75,14 @@ public Observable<String> messages() {
 
         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, ""));
     }
 
     private void createWebSocketConnection() {
@@ -71,9 +93,9 @@ private void createWebSocketConnection() {
 
         Request.Builder requestBuilder = new Request.Builder()
                 .url(mUri);
-        
+
         addConnectionHeadersToBuilder(requestBuilder, mConnectHttpHeaders);
-        
+
         openedSocked = mOkHttpClient.newWebSocket(requestBuilder.build(),
                 new WebSocketListener() {
                     @Override
@@ -113,19 +135,24 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) {
 
     @Override
     public Observable<Void> send(String stompMessage) {
-        return Observable.create(subscriber -> {
+        // .create(onSubscribe) is deprecated because it's unsafe
+        return Observable.fromCallable(() -> {
             if (openedSocked == null) {
-                subscriber.onError(new IllegalStateException("Not connected yet"));
+                throw new IllegalStateException("Not connected yet");
             } else {
                 Log.d(TAG, "Send STOMP message: " + stompMessage);
                 openedSocked.send(stompMessage);
-                subscriber.onCompleted();
+                return null;
             }
         });
     }
 
     @Override
     public Observable<LifecycleEvent> getLifecycleReceiver() {
+        // Once again, opting to leave Subscriber tracking to Rx
+        return mLifecycleStream;
+
+        /*
         return Observable.<LifecycleEvent>create(subscriber -> {
             mLifecycleSubscribers.add(subscriber);
 
@@ -135,6 +162,7 @@ public Observable<LifecycleEvent> getLifecycleReceiver() {
                 if (iterator.next().isUnsubscribed()) iterator.remove();
             }
         });
+        */
     }
 
     private TreeMap<String, String> headersAsMap(Response response) {
@@ -154,15 +182,24 @@ 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

@@ -20,6 +20,7 @@
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLSocketFactory;
 
+import rx.Completable;
 import rx.Observable;
 import rx.Subscriber;
 
@@ -145,6 +146,14 @@ public Observable<Void> send(String stompMessage) {
         });
     }
 
+    // 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) {

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

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