Merge "In the WebSocket client code, a close WebSocket frame was not being sent when the Client was closing the connection causing long hangs. When the Mqtt Client tries to disconnect, the WebSocket client sends the correct opcode to close the connection. Signed-off-by: James Sutton <james.sutton@uk.ibm.com>" into develop

Author: Ian Craggs

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketFrame.java

@@ -27,6 +27,7 @@ public class WebSocketFrame {
 	private byte opcode;
 	private boolean fin;
 	private byte payload[];
+	private boolean closeFlag = false;
 
 	public byte getOpcode() {
 		return opcode;
@@ -39,6 +40,10 @@ public boolean isFin() {
 	public byte[] getPayload() {
 		return payload;
 	}
+	
+	public boolean isCloseFlag() {
+		return closeFlag;
+	}
 
 
 	/**
@@ -127,57 +132,62 @@ private void setFinAndOpCode(byte incomingByte){
 	public WebSocketFrame(InputStream input) throws IOException {
 		byte firstByte = (byte) input.read();
 		setFinAndOpCode(firstByte);
-		if(this.opcode != 2){
-			throw new IOException("Invalid Frame");
-		}
-		
-		byte maskLengthByte = (byte) input.read();
-		boolean masked = ((maskLengthByte & 0x80) != 0);
-		int payloadLength = (byte)(0x7F & maskLengthByte);
-		int byteCount = 0;
-		if(payloadLength == 0X7F){
-			// 8 Byte Extended payload length
-			byteCount = 8;
-		} else if (payloadLength == 0X7E){
-			// 2 bytes extended payload length
-			byteCount = 2;
-		}
-		
-		// Decode the payload length
-		if(byteCount > 0){
-			payloadLength = 0;
-		}
-		while (--byteCount >= 0){
-			maskLengthByte = (byte) input.read();
-			payloadLength |= (maskLengthByte & 0xFF) << (8 * byteCount);
-		}
-		
-		// Get the masking key
-		byte maskingKey[] = null;
-		if(masked) {
-			maskingKey = new byte[4];
-			input.read(maskingKey,0,4);
-		}
+		if(this.opcode == 2){
+			byte maskLengthByte = (byte) input.read();
+			boolean masked = ((maskLengthByte & 0x80) != 0);
+			int payloadLength = (byte)(0x7F & maskLengthByte);
+			int byteCount = 0;
+			if(payloadLength == 0X7F){
+				// 8 Byte Extended payload length
+				byteCount = 8;
+			} else if (payloadLength == 0X7E){
+				// 2 bytes extended payload length
+				byteCount = 2;
+			}
+			
+			// Decode the payload length
+			if(byteCount > 0){
+				payloadLength = 0;
+			}
+			while (--byteCount >= 0){
+				maskLengthByte = (byte) input.read();
+				payloadLength |= (maskLengthByte & 0xFF) << (8 * byteCount);
+			}
+			
+			// Get the masking key
+			byte maskingKey[] = null;
+			if(masked) {
+				maskingKey = new byte[4];
+				input.read(maskingKey,0,4);
+			}
 
-		this.payload = new byte[payloadLength];
-		int offsetIndex = 0;
-		int tempLength = payloadLength;
-		int bytesRead = 0;
-		while (offsetIndex != payloadLength){
-				bytesRead = input.read(this.payload,offsetIndex,tempLength);
-				offsetIndex += bytesRead;
-				tempLength -= bytesRead;
+			this.payload = new byte[payloadLength];
+			int offsetIndex = 0;
+			int tempLength = payloadLength;
+			int bytesRead = 0;
+			while (offsetIndex != payloadLength){
+					bytesRead = input.read(this.payload,offsetIndex,tempLength);
+					offsetIndex += bytesRead;
+					tempLength -= bytesRead;
+			}
+			
+			
+			// Demask if needed
+			if(masked)
+			{
+				for(int i = 0; i < this.payload.length; i++){
+					this.payload[i] ^= maskingKey[i % 4];
+				}
+			}
+			return;
+		} else if(this.opcode == 8){
+			// Closing connection with server
+			closeFlag = true;
+		} else {
+			throw new IOException("Invalid Frame: Opcode: " +this.opcode);
 		}
 
 
-		// Demask if needed
-		if(masked)
-		{
-			for(int i = 0; i < this.payload.length; i++){
-				this.payload[i] ^= maskingKey[i % 4];
-			}
-		}
-		return;
 	}
 
 
@@ -288,6 +298,9 @@ public static byte[] generateMaskingKey(){
 		int d = randomGenerator.nextInt(255);
 		return new byte[] {(byte) a,(byte) b,(byte) c,(byte) d};
 	}
+
+
+
 
 
 }

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketNetworkModule.java

@@ -20,6 +20,7 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.PipedInputStream;
+import java.net.SocketException;
 import java.nio.ByteBuffer;
 
 import javax.net.SocketFactory;
@@ -98,6 +99,12 @@ public OutputStream getOutputStream() throws IOException {
 	 * Stops the module, by closing the TCP socket.
 	 */
 	public void stop() throws IOException {
+		// Creating Close Frame
+		WebSocketFrame frame = new WebSocketFrame((byte)0x08, true, "1000".getBytes());
+		byte[] rawFrame = frame.encodeFrame();
+		getSocketOutputStream().write(rawFrame);
+		getSocketOutputStream().flush();
+		
 		if(webSocketReceiver != null){
 			webSocketReceiver.stop();
 		}

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketReceiver.java

@@ -29,6 +29,7 @@ public class WebSocketReceiver implements Runnable{
 	private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 
 	private boolean running = false;
+	private boolean stopping = false;
 	private Object lifecycle = new Object();
 	private InputStream input;
 	private Thread receiverThread = null;
@@ -63,6 +64,7 @@ public void start(String threadName){
 	 */
 	public void stop() {
 		final String methodName = "stop";
+		stopping = true;
 		synchronized (lifecycle) {
 			//@TRACE 850=stopping
 			log.fine(CLASS_NAME,methodName, "850");
@@ -94,18 +96,27 @@ public void run() {
 				log.fine(CLASS_NAME, methodName, "852");
 				receiving = input.available() > 0;
 				WebSocketFrame incomingFrame = new WebSocketFrame(input);
+				if(!incomingFrame.isCloseFlag()){
 				for(int i = 0; i < incomingFrame.getPayload().length; i++){
 					pipedOutputStream.write(incomingFrame.getPayload()[i]);
 				}
+				
 				pipedOutputStream.flush();
+				} else {
+					if(!stopping){
+						throw new IOException("Server sent a WebSocket Frame with the Stop OpCode");
+					}
+				}
+				
 				receiving = false;
 
 			} catch (IOException ex) {
 				// Exception occurred whilst reading the stream. 
+				System.out.println("WebSocketReceiver.java : run(): Exception Occured.");
+				ex.printStackTrace();
 				closeOutputStream();
 			}
 		}
-		
 	}
 
 	private void closeOutputStream(){