Synchornize websockets writes

jacek-marchwicki · jacek-marchwicki · commit 037276724052 · 2016-02-08T17:33:31.000+01:00
Change-Id: If5b81abe1a161a34479b2115d0918c96414ead54
diff --git a/README.md b/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
diff --git a/websockets-rxjava/build.gradle b/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)
diff --git a/websockets-rxjava/src/main/java/com/appunite/websocket/rx/RxWebSockets.java b/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);
+        }
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ signing {`
`46`	`46`	`}`
`47`	`47`	`group = "com.appunite"`
`48`	`48`	`archivesBaseName = "websockets-rxjava"`
`49`		`-version = "4.0.0"`
	`49`	`+version = "4.0.1"`
`50`	`50`
`51`	`51`	`if (!project.hasProperty("ossrhUsername")) {`
`52`	`52`	`project.ext.setProperty("ossrhUsername", null)`