Merge pull request #500 from marci4/master

Making WebSocket.send() thread-safe

Author: marci4

pom.xml

@@ -16,12 +16,6 @@
   <scm>
     <url>https://github.com/TooTallNate/Java-WebSocket</url>
   </scm>
-  <distributionManagement>
-    <snapshotRepository>
-      <id>ossrh</id>
-      <url>https://oss.sonatype.org/content/repositories/snapshots</url>
-    </snapshotRepository>
-  </distributionManagement>
   <build>
     <sourceDirectory>src/main/java</sourceDirectory>
     <plugins>
@@ -62,6 +56,20 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-gpg-plugin</artifactId>
+        <version>1.5</version>
+        <executions>
+          <execution>
+            <id>sign-artifacts</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>sign</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
   <dependencies>

src/main/java/org/java_websocket/WebSocketImpl.java

@@ -25,6 +25,7 @@
 import java.nio.channels.SelectionKey;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -110,6 +111,11 @@ public class WebSocketImpl implements WebSocket {
 	 */
 	private long lastPong = System.currentTimeMillis();
 
+	/**
+	 * Attribut to synchronize the write
+	 */
+	private static final Object synchronizeWriteObject = new Object();
+
 	/**
 	 * Creates a websocket with server role
 	 *
@@ -209,8 +215,8 @@ private boolean decodeHandshake( ByteBuffer socketBufferNew ) {
 				HandshakeState isflashedgecase = isFlashEdgeCase( socketBuffer );
 				if( isflashedgecase == HandshakeState.MATCHED ) {
 					try {
-						write( ByteBuffer.wrap( Charsetfunctions.utf8Bytes( wsl.getFlashPolicy( this ) ) ) );
-						close( CloseFrame.FLASHPOLICY, "" );
+						write( Collections.singletonList( ByteBuffer.wrap(Charsetfunctions.utf8Bytes(wsl.getFlashPolicy(this)))));
+						close(CloseFrame.FLASHPOLICY, "");
 					} catch ( InvalidDataException e ) {
 						close( CloseFrame.ABNORMAL_CLOSE, "remote peer closed connection before flashpolicy could be transmitted", true );
 					}
@@ -623,9 +629,13 @@ public void send( byte[] bytes ) throws IllegalArgumentException, WebsocketNotCo
 	private void send( Collection<Framedata> frames ) {
 		if( !isOpen() )
 			throw new WebsocketNotConnectedException();
-		for( Framedata f : frames ) {
-			sendFrame( f );
+		ArrayList<ByteBuffer> outgoingFrames = new ArrayList<ByteBuffer>();
+		for (Framedata f : frames) {
+			if( DEBUG )
+				System.out.println( "send frame: " + f );
+			 outgoingFrames.add( draft.createBinaryFrame( f ) );
 		}
+		write( outgoingFrames );
 	}
 
 	@Override
@@ -635,9 +645,7 @@ public void sendFragmentedFrame( Opcode op, ByteBuffer buffer, boolean fin ) {
 
 	@Override
 	public void sendFrame( Framedata framedata ) {
-		if( DEBUG )
-			System.out.println( "send frame: " + framedata );
-		write( draft.createBinaryFrame( framedata ) );
+		send ( Collections.singletonList( framedata ) );
 	}
 
 	public void sendPing() throws NotYetConnectedException {
@@ -707,8 +715,10 @@ private void write( ByteBuffer buf ) {
 	}
 
 	private void write( List<ByteBuffer> bufs ) {
-		for( ByteBuffer b : bufs ) {
-			write( b );
+		synchronized ( synchronizeWriteObject ) {
+			for (ByteBuffer b : bufs) {
+				write(b);
+			}
 		}
 	}