diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java index d037fdd3..6553a74f 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java +++ b/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 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 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)); diff --git a/tools_scripts/webSocketClient.groovy b/tools_scripts/webSocketClient.groovy new file mode 100644 index 00000000..2f992023 --- /dev/null +++ b/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() diff --git a/tools_scripts/web_socket_test.html b/tools_scripts/web_socket_test.html new file mode 100644 index 00000000..05c36704 --- /dev/null +++ b/tools_scripts/web_socket_test.html @@ -0,0 +1,101 @@ + + + + + + + + + + + + +

Example Web Messaging web page.

+
+
+ Connect + Make a connection to the server, and set up a call back used if a + message arrives for this client. +
+ + +
+ +
+ Subscribe + Make a subscription to topic "/World". +
+
+ +
+ Send + Create a Message object containing the word "Hello" and then publish it at + the server. +
+ +
+ +
+ Receive + A copy of the published Message is received in the callback we created earlier. + +
+ +
+ Disconnect + Now disconnect this client from the server. +
+
+
+ +