diff --git a/broker/config/moquette.conf b/broker/config/moquette.conf index d50351c7..0835f78e 100644 --- a/broker/config/moquette.conf +++ b/broker/config/moquette.conf @@ -7,6 +7,8 @@ port 1883 +websocket_port 8080 + host 0.0.0.0 password_file config/password_file.conf diff --git a/broker/pom.xml b/broker/pom.xml index 93a3868f..d5d17aa1 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -99,6 +99,12 @@ 4.0.8.Final + + io.netty + netty-codec-http + 4.0.8.Final + + org.fusesource.hawtbuf hawtbuf @@ -139,6 +145,13 @@ 0.4.0 test + + + org.eclipse.jetty.websocket + websocket-client + 9.2.0.M1 + test + diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java index 845bbee4..eb3f9c4f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java @@ -29,6 +29,7 @@ class ConfigurationParser { ConfigurationParser() { m_properties.put("port", Integer.toString(Constants.PORT)); m_properties.put("host", Constants.HOST); + m_properties.put("websocket_port", Integer.toString(Constants.WEBSOCKET_PORT)); m_properties.put("password_file", ""); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index b207d96b..bf273583 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -26,7 +26,7 @@ public class Server { final Server server = new Server(); server.startServer(); - System.out.println("Server started, version 0.5"); + System.out.println("Server started, version 0.6-SNAPSHOT"); //Bind a shutdown hook Runtime.getRuntime().addShutdownHook(new Thread() { @Override diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java index f0ad2a2d..ee529a3b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -1,7 +1,9 @@ package org.dna.mqtt.moquette.server.netty; import io.netty.bootstrap.ServerBootstrap; +import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; @@ -9,10 +11,18 @@ import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.MessageToMessageDecoder; +import io.netty.handler.codec.MessageToMessageEncoder; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpRequestDecoder; +import io.netty.handler.codec.http.HttpResponseEncoder; +import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; +import io.netty.handler.timeout.IdleStateHandler; import java.io.IOException; -import io.netty.handler.timeout.IdleStateHandler; +import java.util.List; import java.util.Properties; import org.dna.mqtt.commons.Constants; import org.dna.mqtt.moquette.messaging.spi.IMessaging; @@ -29,42 +39,73 @@ import org.slf4j.LoggerFactory; */ public class NettyAcceptor implements ServerAcceptor { + static class WebSocketFrameToByteBufDecoder extends MessageToMessageDecoder { + + @Override + protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List out) throws Exception { + //convert the frame to a ByteBuf + ByteBuf bb = frame.content(); + out.add(bb); + } + } + + static class ByteBufToWebSocketFrameEncoder extends MessageToMessageEncoder { + + @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); + } + } + private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class); EventLoopGroup m_bossGroup; EventLoopGroup m_workerGroup; //BytesMetricsCollector m_metricsCollector = new BytesMetricsCollector(); MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector(); + MQTTEncoder mqttEncoder = new MQTTEncoder(); + MQTTDecoder mqttDecoder = new MQTTDecoder(); + @Override public void initialize(IMessaging messaging, Properties props) throws IOException { - m_bossGroup = new NioEventLoopGroup(); + m_bossGroup = new NioEventLoopGroup(); m_workerGroup = new NioEventLoopGroup(); + initializePlainTCPTransport(messaging, props); + initializeWebSocketTransport(messaging, props); + } + + private void initializePlainTCPTransport(IMessaging messaging, Properties props) throws IOException { +// m_bossGroup = new NioEventLoopGroup(); +// m_workerGroup = new NioEventLoopGroup(); + final NettyMQTTHandler handler = new NettyMQTTHandler(); handler.setMessaging(messaging); ServerBootstrap b = new ServerBootstrap(); - b.group(m_bossGroup, m_workerGroup) - .channel(NioServerSocketChannel.class) - .childHandler(new ChannelInitializer() { - @Override - public void initChannel(SocketChannel ch) throws Exception { - ChannelPipeline pipeline = ch.pipeline(); + b.group(m_bossGroup, m_workerGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + ChannelPipeline pipeline = ch.pipeline(); //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector)); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); - //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); - pipeline.addLast("decoder", new MQTTDecoder()); - pipeline.addLast("encoder", new MQTTEncoder()); - pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); - pipeline.addLast("handler", handler); - } - }) - .option(ChannelOption.SO_BACKLOG, 128) - .option(ChannelOption.SO_REUSEADDR, true) - .childOption(ChannelOption.SO_KEEPALIVE, true); - try { + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); + pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); + pipeline.addLast("decoder", new MQTTDecoder()); + pipeline.addLast("encoder", new MQTTEncoder()); + pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); + pipeline.addLast("handler", handler); + } + }) + .option(ChannelOption.SO_BACKLOG, 128) + .option(ChannelOption.SO_REUSEADDR, true) + .childOption(ChannelOption.SO_KEEPALIVE, true); + try { // Bind and start to accept incoming connections. // ChannelFuture f = b.bind(Constants.PORT); ChannelFuture f = b.bind(props.getProperty("host"), Integer.parseInt(props.getProperty("port"))); @@ -74,6 +115,48 @@ public class NettyAcceptor implements ServerAcceptor { LOG.error(null, ex); } } + + private void initializeWebSocketTransport(IMessaging messaging, Properties props) throws IOException { +// m_bossGroup = new NioEventLoopGroup(); +// m_workerGroup = new NioEventLoopGroup(); + + final NettyMQTTHandler handler = new NettyMQTTHandler(); + handler.setMessaging(messaging); + + ServerBootstrap b = new ServerBootstrap(); + b.group(m_bossGroup, m_workerGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + + ChannelPipeline pipeline = ch.pipeline(); + + pipeline.addLast("httpEncoder", new HttpResponseEncoder()); + pipeline.addLast("httpDdecoder", new HttpRequestDecoder()); + pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); + pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqttv3.1")); + pipeline.addLast("bytebuf2wsDecoderr", new ByteBufToWebSocketFrameEncoder()); + pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); + pipeline.addLast("decoder", new MQTTDecoder()); + pipeline.addLast("encoder", new MQTTEncoder()); + pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); + pipeline.addLast("handler", handler); + } + }) + .option(ChannelOption.SO_BACKLOG, 128) + .option(ChannelOption.SO_REUSEADDR, true) + .childOption(ChannelOption.SO_KEEPALIVE, true); + try { + int port = Integer.parseInt(props.getProperty("websocket_port")); + ChannelFuture f = b.bind(props.getProperty("host"), port); + System.out.println("Web socket server started at port " + port + '.'); + System.out.println("Open your browser and navigate to http://localhost:" + port + '/'); + f.sync(); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } public void close() { if (m_workerGroup == null) { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/MQTTWebSocket.java b/broker/src/test/java/org/dna/mqtt/moquette/server/MQTTWebSocket.java new file mode 100644 index 00000000..be29c043 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/MQTTWebSocket.java @@ -0,0 +1,83 @@ +package org.dna.mqtt.moquette.server; + +import java.io.IOException; +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.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("Connection closed: %d - %s%n", statusCode, reason); + this.session = null; + } + + @OnWebSocketConnect + public void onConnect(Session session) throws IOException { + System.out.printf("Got connect: %s%n", session); + LOG.info("Got connect: %s%n", session); + 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 + + ByteBuffer msg = ByteBuffer.wrap(new byte[] { 0x10, //message type + 0x0C, //remaining lenght 12 bytes + 'M', 'Q','I','s','d','p', + 0x03, //protocol version + 0x32, //flags + 0x02, 0x00// keepAlive Timer, 2 seconds + }); +// try { + session.getRemote().sendBytes(msg); + this.connectSentLatch.countDown(); + session.close(StatusCode.NORMAL, "I'm done"); +// } catch (IOException t) { +// LOG.error(null, t); +// } + } + + public boolean awaitConnected(int duration, TimeUnit unit) throws InterruptedException { + return this.connectSentLatch.await(duration, unit); + } + + @OnWebSocketMessage + public void onMessage(Session session, byte[] buf, int offset, int length) { + System.out.printf("Got msg: %s%n", buf); + } + + @OnWebSocketError + public void onError(Session session, Throwable cause) { + System.out.printf("Got exception: %s%n", cause); + } +} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index 26820e23..9e5eb876 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -64,7 +64,7 @@ public class ServerIntegrationFuseTest { } @Test - public void connectWithCredetials() throws Exception { + public void connectWithCredentials() throws Exception { LOG.info("*** connectWithCredetials ***"); MQTT mqtt = new MQTT(); mqtt.setHost("localhost", 1883); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationWebSocketTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationWebSocketTest.java new file mode 100644 index 00000000..fb64f858 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationWebSocketTest.java @@ -0,0 +1,66 @@ +package org.dna.mqtt.moquette.server; + +import java.io.File; +import java.io.IOException; +import java.net.URI; +import java.util.concurrent.TimeUnit; +import org.eclipse.jetty.websocket.client.WebSocketClient; +import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; +import org.dna.mqtt.commons.Constants; +import org.junit.After; +import static org.junit.Assert.assertTrue; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Integration test to check the function of Moquette with a WebSocket channel. + * + * @author andrea + */ +public class ServerIntegrationWebSocketTest { + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); + + Server m_server; + WebSocketClient client; + + protected void startServer() throws IOException { + m_server = new Server(); + m_server.startServer(); + } + + @Before + public void setUp() throws Exception { + startServer(); + client = new WebSocketClient(); + } + + @After + public void tearDown() throws Exception { + client.stop(); + + m_server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + } + + @Test + public void checkPlainConnect() throws Exception { + LOG.info("*** checkPlainConnect ***"); + String destUri = "ws://localhost:" + Constants.WEBSOCKET_PORT; + + MQTTWebSocket socket = new MQTTWebSocket(); + client.start(); + URI echoUri = new URI(destUri); + ClientUpgradeRequest request = new ClientUpgradeRequest(); + client.connect(socket, echoUri, request); + LOG.info("Connecting to : %s%n", echoUri); + boolean connected = socket.awaitConnected(4, TimeUnit.SECONDS); + + assertTrue(connected); + } +} diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index 1f7a600c..001f5889 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -7,7 +7,9 @@ port 1883 +websocket_port 8080 + host 0.0.0.0 #Password file -password_file config/password_file.conf \ No newline at end of file +password_file config/password_file.conf diff --git a/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java b/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java index 5130dc22..780e1fbc 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java +++ b/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java @@ -5,6 +5,7 @@ package org.dna.mqtt.commons; */ public class Constants { public static final int PORT = 1883; + public static final int WEBSOCKET_PORT = 8080; public static final String HOST = "0.0.0.0"; public static final int DEFAULT_CONNECT_TIMEOUT = 10; }