From e71248f169682e528302111ab276174263dfa278 Mon Sep 17 00:00:00 2001
From: Nickolay Savchenko <savchenko@river-soft.net>
Date: Thu, 7 Feb 2019 18:49:19 +0200
Subject: [PATCH] Fix unsubscribe

---
 lib/src/main/java/ua/naiksoftware/stomp/StompClient.java | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/lib/src/main/java/ua/naiksoftware/stomp/StompClient.java b/lib/src/main/java/ua/naiksoftware/stomp/StompClient.java
index 3b2be91..a16ecde 100644
--- a/lib/src/main/java/ua/naiksoftware/stomp/StompClient.java
+++ b/lib/src/main/java/ua/naiksoftware/stomp/StompClient.java
@@ -174,11 +174,7 @@ public Completable send(String destination, String data) {
     public Completable send(@NonNull StompMessage stompMessage) {
         Completable completable = connectionProvider.send(stompMessage.compile(legacyWhitespace));
         CompletableSource connectionComplete = getConnectionStream()
-//                .filter(isConnected -> isConnected)
-                .filter(isConnected -> {
-                    Log.d(TAG, "Send " + stompMessage + "filtered " + isConnected);
-                    return isConnected;
-                })
+                .filter(isConnected -> isConnected)
                 .firstElement().ignoreElement();
         return completable
                 .startWith(connectionComplete);
@@ -248,8 +244,7 @@ else if (!streamMap.containsKey(destPath))
                     getMessageStream()
                             .filter(msg -> pathMatcher.matches(destPath, msg))
                             .toFlowable(BackpressureStrategy.BUFFER)
-                            .doFinally(() -> unsubscribePath(destPath).subscribe())
-                            .share())
+                            .share()).doFinally(() -> unsubscribePath(destPath).subscribe())
             );
         return streamMap.get(destPath);
     }