Fixed issue #5 and websocket implementation in general

broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java

@@ -17,6 +17,7 @@ package org.eclipse.moquette.server.netty;
 
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
@@ -66,6 +67,8 @@ public class NettyAcceptor implements ServerAcceptor {
         protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List<Object> out) throws Exception {
             //convert the frame to a ByteBuf
             ByteBuf bb = frame.content();
+            System.out.println("WebSocketFrameToByteBufDecoder decode - " + ByteBufUtil.hexDump(bb));
+            bb.retain();
             out.add(bb);
         }
     }
@@ -75,8 +78,10 @@ public class NettyAcceptor implements ServerAcceptor {
         @Override
         protected void encode(ChannelHandlerContext chc, ByteBuf bb, List<Object> out) throws Exception {
             //convert the ByteBuf to a WebSocketFrame
-            BinaryWebSocketFrame result = new BinaryWebSocketFrame(bb);
-            out.add(bb);
+            BinaryWebSocketFrame result = new BinaryWebSocketFrame();
+            System.out.println("ByteBufToWebSocketFrameEncoder encode - " + ByteBufUtil.hexDump(bb));
+            result.content().writeBytes(bb);
+            out.add(result);
         }
     }
 
@@ -162,11 +167,14 @@ public class NettyAcceptor implements ServerAcceptor {
             @Override
             void init(ChannelPipeline pipeline) {
                 pipeline.addLast("httpEncoder", new HttpResponseEncoder());
-                pipeline.addLast("httpDdecoder", new HttpRequestDecoder());
+                pipeline.addLast("httpDecoder", new HttpRequestDecoder());
                 pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
-                pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqttv3.1"));
-                pipeline.addLast("bytebuf2wsDecoderr", new ByteBufToWebSocketFrameEncoder());
+                pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt"/*"/mqtt"*/, "mqttv3.1, mqttv3.1.1"));
+                //pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler(null, "mqtt"));
                 pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
+                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
+                pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
+                pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
                 pipeline.addLast("decoder", new MQTTDecoder());
                 pipeline.addLast("encoder", new MQTTEncoder());
                 pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));

tools_scripts/webSocketClient.groovy

@@ -0,0 +1,121 @@
+@Grab(group='org.eclipse.jetty.websocket', module='websocket-client', version='9.2.0.M1', ext='jar')
+@Grab(group='org.slf4j', module='slf4j-api', version='1.7.7')
+@Grab(group='org.slf4j', module='slf4j-simple', version='1.7.7')
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.StatusCode;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
+import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author andrea
+ */
+@WebSocket
+public class MQTTWebSocket {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MQTTWebSocket.class);
+
+    private Session session;
+
+    private final CountDownLatch connectSentLatch;
+
+    public MQTTWebSocket() {
+        this.connectSentLatch = new CountDownLatch(1);
+    }
+
+    @OnWebSocketClose
+    public void onClose(int statusCode, String reason) {
+        System.out.printf("onClose - Connection closed: %d - %s%n", statusCode, reason);
+        this.session = null;
+    }
+
+    @OnWebSocketConnect
+    public void onConnect(Session session) throws IOException {
+        System.out.println("onConnect - Got connect");
+        this.session = session;
+
+        //Trivial MQTT Connect message
+//        assertEquals(12, out.readByte()); //remaining length
+//        verifyString("MQIsdp", out);
+//        assertEquals(0x03, out.readByte()); //protocol version
+//        assertEquals(0x32, out.readByte()); //flags
+//        assertEquals(2, out.readByte()); //keepAliveTimer msb
+//        assertEquals(0, out.readByte()); //keepAliveTimer lsb
+
+        byte[] rawMsg = [ 0x10, //message type
+                          0x11, //remaining length 17 bytes
+                          0x00, 0x06, //MQIspd length
+                          'M' as char, 'Q' as char, 'I' as char, 's' as char,'d' as char, 'p' as char,
+                          0x03, //protocol version
+                          0x32, //flags
+                          0x00, 0x0A,// keepAlive Timer, 10 seconds
+                          0x00, 0x03, 0x41, 0x42, 0x43 // A, B, C as client ID
+        ] as byte[]
+        ByteBuffer msg = ByteBuffer.wrap(rawMsg);
+
+        //DBG
+        println "onConnect - send bytes " + rawMsg.encodeHex().toString()
+        //DBG
+
+//        try {
+        session.getRemote().sendBytes(msg);
+        println("onConnect - Sent connect bytes");
+        this.connectSentLatch.countDown();
+        //session.close(StatusCode.NORMAL, "I'm done");
+        println("onConnect - After connect");
+
+        println "send disconnect"
+        disconnect()
+    }
+
+    public boolean awaitConnected(int duration, TimeUnit unit) throws InterruptedException {
+        return this.connectSentLatch.await(duration, unit);
+    }
+
+    /**
+     * Send a disconnect message to the server
+     * */
+    public void disconnect() {
+        byte[] rawMsg = [0xE0, 0x00] as byte[]
+        println "disconnect - send bytes " + rawMsg.encodeHex().toString()
+        ByteBuffer msg = ByteBuffer.wrap(rawMsg);
+        this.session.getRemote().sendBytes(msg)
+    }
+
+    @OnWebSocketMessage
+    public void onMessage(Session session, byte[] buf, int offset, int length) {
+        System.out.printf("onMessage - Got msg: %s%n", buf);
+    }
+
+    @OnWebSocketError
+    public void onError(Session session, Throwable cause) {
+        System.out.printf("onError - Got exception: %s%n", cause);
+    }
+}
+
+
+String destUri = "ws://localhost:8080/mqtt"
+//String destUri = "ws://test.mosquitto.org:8080/"
+WebSocketClient client = new WebSocketClient()
+MQTTWebSocket socket = new MQTTWebSocket();
+client.start();
+URI echoUri = new URI(destUri);
+ClientUpgradeRequest request = new ClientUpgradeRequest();
+client.connect(socket, echoUri, request);
+println "main - Connecting to : ${echoUri}"
+boolean connected = socket.awaitConnected(4, TimeUnit.SECONDS);
+println "main - After wait for awaitConnected was ${connected}"
+
+//println "send disconnect"
+//socket.disconnect()

tools_scripts/web_socket_test.html

@@ -0,0 +1,101 @@
+<!DOCTYPE html>
+
+
+<head>
+  <!--script type="text/javascript" src="paho.javascript-1.0.0/mqttws31.js"></script-->
+  <script type="application/javascript" src="http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt.javascript.git/plain/src/mqttws31.js"></script>
+  <!--script type="text/javascript" src="file:///home/andrea/workspace/moquette_github/mqttws31.js"></script-->
+
+  <script type="text/javascript">
+  
+  // Create a client instance
+        var client;
+        function doConnect() {
+                console.log("connect invoked");
+                client = new Paho.MQTT.Client("127.0.0.1", Number(8080), "", "clientId");
+                //client = new Paho.MQTT.Client("test.mosquitto.org", Number(8080), "", "clientId");
+                //client = new Paho.MQTT.Client("ws://whistler1.hursley.ibm.com", Number(1883), "ClientId");
+
+                client.startTrace();
+                // set callback handlers
+                client.onConnectionLost = onConnectionLost;
+                client.onMessageArrived = onMessageArrived;
+
+                // connect the client
+                client.connect({onSuccess:onConnect});
+                console.log("connect finished");
+        }
+
+
+        // called when the client connects
+        function onConnect() {
+          // Once a connection has been made, make a subscription and send a message.
+          console.log("onConnect");
+          client.subscribe("/World");
+          message = new Paho.MQTT.Message("Hello");
+          message.destinationName = "/World";
+          client.send(message); 
+        }
+
+        // called when the client loses its connection
+        function onConnectionLost(responseObject) {
+          if (responseObject.errorCode !== 0) {
+            console.log("onConnectionLost:"+responseObject.errorMessage);
+          }
+        }
+
+        // called when a message arrives
+        function onMessageArrived(message) {
+          console.log("onMessageArrived:"+message.payloadString);
+        } 
+
+        function doSubscribe() {
+              client.subscribe("/World");
+        }
+
+        function doDisconnect() {
+            client.disconnect();
+        }
+  </script>
+</head> 
+
+<body>
+  <h1>Example Web Messaging web page.</h1>
+  <form id="example">
+  <fieldset>
+  <legend id="Connect" > Connect </legend>
+    Make a connection to the server, and set up a call back used if a 
+    message arrives for this client.
+    <br>
+    <input type="button" value="Connect" onClick="doConnect(this.form)" name="Connect"/>
+    <input type="checkbox" name="connected" disabled="disabled"/>
+  </fieldset>
+  
+  <fieldset>
+  <legend id="Subscribe" > Subscribe </legend>
+  Make a subscription to topic "/World".
+  <br> <input type="button" value="Subscribe" onClick="doSubscribe(this.form)"/>
+  </fieldset>
+  
+  <fieldset>
+  <legend id="Send" > Send </legend>
+    Create a Message object containing the word "Hello" and then publish it at
+    the server.
+    <br>
+    <input type="button" value="Send" onClick="doSend(this.form)"/>
+  </fieldset>
+  
+  <fieldset>
+  <legend id="Receive" > Receive </legend>
+    A copy of the published Message is received in the callback we created earlier.
+  <textarea name="receiveMsg" rows="1" cols="40" disabled="disabled"></textarea>
+  </fieldset>
+  
+  <fieldset>
+  <legend id="Disconnect" > Disconnect </legend>
+    Now disconnect this client from the server.
+  <br> <input type="button" value="Disconnect" onClick="doDisconnect()"/>
+  </fieldset>
+  </form>
+</body>
+</html>