Fix memory leaks (clear subscriptions)

Author: NaikSoftware

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

@@ -46,7 +46,7 @@
     @Override
     public Flowable<String> messages() {
         Flowable<String> flowable = Flowable.<String>create(mMessagesEmitters::add, BackpressureStrategy.BUFFER)
-                .doOnCancel(() -> {
+                .doFinally(() -> {
                     Iterator<FlowableEmitter<? super String>> iterator = mMessagesEmitters.iterator();
                     while (iterator.hasNext()) {
                         if (iterator.next().isCancelled()) iterator.remove();
@@ -131,7 +131,7 @@ public Flowable<Void> send(String stompMessage) {
     @Override
     public Flowable<LifecycleEvent> getLifecycleReceiver() {
         return Flowable.<LifecycleEvent>create(mLifecycleEmitters::add, BackpressureStrategy.BUFFER)
-                .doOnCancel(() -> {
+                .doFinally(() -> {
                     synchronized (mLifecycleEmitters) {
                         Iterator<FlowableEmitter<? super LifecycleEvent>> iterator = mLifecycleEmitters.iterator();
                         while (iterator.hasNext()) {

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

@@ -58,7 +58,7 @@
     @Override
     public Flowable<String> messages() {
         Flowable<String> flowable = Flowable.<String>create(mMessagesEmitters::add, BackpressureStrategy.BUFFER)
-                .doOnCancel(() -> {
+                .doFinally(() -> {
                     Iterator<FlowableEmitter<? super String>> iterator = mMessagesEmitters.iterator();
                     while (iterator.hasNext()) {
                         if (iterator.next().isCancelled()) iterator.remove();
@@ -165,7 +165,7 @@ private void emitMessage(String stompMessage) {
     @Override
     public Flowable<LifecycleEvent> getLifecycleReceiver() {
         return Flowable.<LifecycleEvent>create(mLifecycleEmitters::add, BackpressureStrategy.BUFFER)
-                .doOnCancel(() -> {
+                .doFinally(() -> {
                     synchronized (mLifecycleLock) {
                         Iterator<FlowableEmitter<? super LifecycleEvent>> iterator = mLifecycleEmitters.iterator();
                         while (iterator.hasNext()) {

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

@@ -35,7 +35,7 @@ public class StompClient {
 
     private Disposable mMessagesDisposable;
     private Disposable mLifecycleDisposable;
-    private Map<String, Set<FlowableEmitter<? super StompMessage>>> mEmitters = Collections.synchronizedMap(new HashMap<>());
+    private final Map<String, Set<FlowableEmitter<? super StompMessage>>> mEmitters = Collections.synchronizedMap(new HashMap<>());
     private List<ConnectableFlowable<Void>> mWaitConnectionFlowables;
     private final ConnectionProvider mConnectionProvider;
     private HashMap<String, String> mTopics;
@@ -179,7 +179,7 @@ public Flowable<StompMessage> topic(String destinationPath, List<StompHeader> he
                emittersSet.add(emitter);
            }
        }, BackpressureStrategy.BUFFER)
-           .doOnCancel(() -> {
+           .doFinally(() -> {
                synchronized (mEmitters) {
                    Iterator<String> mapIterator = mEmitters.keySet().iterator();
                    while (mapIterator.hasNext()) {