Merge branch 'lock-websocket-write' into 'master'

Synchornize websockets writes



See merge request !6

Author: jacek-marchwicki

README.md

@@ -100,7 +100,7 @@ To run example first run [websocket server](websockets-server/README.md), than u
 to your gradle file:
 
 ```groovy
-compile "com.appunite:websockets-rxjava:4.0.0"
+compile "com.appunite:websockets-rxjava:4.0.1"
 ```
 		
 ## License

websockets-rxjava/build.gradle

@@ -46,7 +46,7 @@ signing {
 }
 group = "com.appunite"
 archivesBaseName = "websockets-rxjava"
-version = "4.0.0"
+version = "4.0.1"
 
 if (!project.hasProperty("ossrhUsername")) {
     project.ext.setProperty("ossrhUsername", null)

websockets-rxjava/src/main/java/com/appunite/websocket/rx/RxWebSockets.java

@@ -25,6 +25,7 @@
 import com.appunite.websocket.rx.messages.RxEventStringMessage;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
+import okhttp3.RequestBody;
 import okhttp3.Response;
 import okhttp3.ResponseBody;
 import okhttp3.ws.WebSocket;
@@ -62,7 +63,6 @@ public RxWebSockets(@Nonnull OkHttpClient client, @Nonnull Request request) {
         this.request = request;
     }
 
-
     /**
      * Returns observable that connected to a websocket and returns {@link RxObjectEvent}'s
      *
@@ -91,7 +91,7 @@ public void onOpen(WebSocket webSocket, Response response) {
                                     subscriber.onNext(new RxEventDisconnected(e));
                                 }
                             } else {
-                                notifyConnected = webSocket;
+                                notifyConnected = new LockingWebSocket(webSocket);
                             }
                             webSocketItem = notifyConnected;
                         }
@@ -178,4 +178,35 @@ public void call() {
             }
         });
     }
+
+    /**
+     * Class that synchronizes writes to websocket
+     */
+    private static class LockingWebSocket implements WebSocket {
+        @Nonnull
+        private final WebSocket webSocket;
+
+        public LockingWebSocket(@Nonnull WebSocket webSocket) {
+            this.webSocket = webSocket;
+        }
+
+        @Override
+        public void sendMessage(RequestBody message) throws IOException {
+            synchronized (this) {
+                webSocket.sendMessage(message);
+            }
+        }
+
+        @Override
+        public void sendPing(Buffer payload) throws IOException {
+            synchronized (this) {
+                webSocket.sendPing(payload);
+            }
+        }
+
+        @Override
+        public void close(int code, String reason) throws IOException {
+            webSocket.close(code, reason);
+        }
+    }
 }