Merge remote-tracking branch 'NaikSoftware/master' into NoJack

# Conflicts:
#	example-client/build.gradle
#	example-client/src/main/java/ua/naiksoftware/stompclientexample/MainActivity.java
#	lib/build.gradle
#	lib/src/main/java/ua/naiksoftware/stomp/ConnectionProvider.java
#	lib/src/main/java/ua/naiksoftware/stomp/OkHttpConnectionProvider.java
#	lib/src/main/java/ua/naiksoftware/stomp/WebSocketsConnectionProvider.java
#	lib/src/main/java/ua/naiksoftware/stomp/client/StompClient.java

Author: forresthopkinsa

.idea/misc.xml

@@ -1,14 +1,14 @@
 package ua.naiksoftware.stompclientexample;
 
+import io.reactivex.Flowable;
 import retrofit2.http.POST;
 import retrofit2.http.Query;
-import rx.Observable;
 
 /**
  * Created by Naik on 24.02.17.
  */
 public interface ExampleRepository {
 
     @POST("hello-convert-and-send")
-    Observable<Void> sendRestEcho(@Query("msg") String message);
+    Flowable<Void> sendRestEcho(@Query("msg") String message);
 }

example-client/src/main/java/ua/naiksoftware/stompclientexample/ExampleRepository.java

@@ -1,7 +1,7 @@
 package ua.naiksoftware.stompclientexample;
 
 import retrofit2.Retrofit;
-import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory;
+import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory;
 import retrofit2.converter.gson.GsonConverterFactory;
 
 /**
@@ -33,7 +33,7 @@ public static RestClient getInstance() {
     private RestClient() {
         Retrofit retrofit = new Retrofit.Builder().baseUrl("http://" + ANDROID_EMULATOR_LOCALHOST + ":" + SERVER_PORT + "/")
                 .addConverterFactory(GsonConverterFactory.create())
-                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
+                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                 .build();
         mExampleRepository = retrofit.create(ExampleRepository.class);
     }

example-client/src/main/java/ua/naiksoftware/stompclientexample/RestClient.java

@@ -82,6 +82,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/OkHttpConnectionProvider.java

@@ -6,9 +6,9 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 
 import java8.util.StringJoiner;
 import java8.util.concurrent.CompletableFuture;
@@ -33,14 +33,14 @@ public class StompClient {
 
     private final String tag = StompClient.class.getSimpleName();
     private final ConnectionProvider mConnectionProvider;
-    private HashMap<String, String> mTopics;
+    private ConcurrentHashMap<String, String> mTopics;
     private boolean mConnected;
     private boolean isConnecting;
 
     private PublishSubject<StompMessage> mMessageStream;
     private CompletableFuture<Boolean> mConnectionFuture;
     private Completable mConnectionComplete;
-    private HashMap<String, Observable<StompMessage>> mStreamMap;
+    private ConcurrentHashMap<String, Observable<StompMessage>> mStreamMap;
     private Parser parser;
     private Subscription lifecycleSub;
     private List<StompHeader> mHeaders;
@@ -49,7 +49,7 @@ public class StompClient {
     public StompClient(ConnectionProvider connectionProvider) {
         mConnectionProvider = connectionProvider;
         mMessageStream = PublishSubject.create();
-        mStreamMap = new HashMap<>();
+        mStreamMap = new ConcurrentHashMap<>();
         resetStatus();
         parser = Parser.NONE;
     }
@@ -252,7 +252,7 @@ private boolean matches(String path, StompMessage msg) {
     private Completable subscribePath(String destinationPath, @Nullable List<StompHeader> headerList) {
         String topicId = UUID.randomUUID().toString();
 
-        if (mTopics == null) mTopics = new HashMap<>();
+        if (mTopics == null) mTopics = new ConcurrentHashMap<>();
 
         // Only continue if we don't already have a subscription to the topic
         if (mTopics.containsKey(destinationPath)) {