Merge branch 'master' into npe-on-dispose

Author: NaikSoftware

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

@@ -106,6 +106,11 @@ public void onClosed(WebSocket webSocket, int code, String reason) {
                     public void onFailure(WebSocket webSocket, Throwable t, Response response) {
                         emitLifecycleEvent(new LifecycleEvent(LifecycleEvent.Type.ERROR, new Exception(t)));
                     }
+                    
+                    @Override
+                    public void onClosing(final WebSocket webSocket, final int code, final String reason) {
+                        webSocket.close(code, reason);
+                    }
                 }
 
         );

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

@@ -34,6 +34,7 @@ public class StompClient {
     public static final String DEFAULT_ACK = "auto";
 
     private Disposable mMessagesDisposable;
+    private Disposable mLifecycleDisposable;
     private Map<String, Set<FlowableEmitter<? super StompMessage>>> mEmitters = new ConcurrentHashMap<>();
     private List<ConnectableFlowable<Void>> mWaitConnectionFlowables;
     private final ConnectionProvider mConnectionProvider;
@@ -74,7 +75,7 @@ public void connect(List<StompHeader> _headers) {
     public void connect(List<StompHeader> _headers, boolean reconnect) {
         if (reconnect) disconnect();
         if (mConnected) return;
-        mConnectionProvider.getLifecycleReceiver()
+        mLifecycleDisposable = mConnectionProvider.getLifecycleReceiver()
                 .subscribe(lifecycleEvent -> {
                     switch (lifecycleEvent.getType()) {
                         case OPENED:
@@ -156,6 +157,7 @@ public Flowable<LifecycleEvent> lifecycle() {
 
     public void disconnect() {
         if (mMessagesDisposable != null) mMessagesDisposable.dispose();
+        if (mLifecycleDisposable != null) mLifecycleDisposable.dispose();
         mConnected = false;
     }