anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added websocket connector

Browse Source
This commit is contained in:
selva.andre
2014-05-18 12:02:18 +02:00
parent 17548eeff7
commit 6755723e0b
10 changed files with 275 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/config/moquette.conf
+2
View File
@@ -7,6 +7,8 @@
port 1883
websocket_port 8080
host 0.0.0.0
password_file config/password_file.conf
broker/pom.xml
+13
View File
@@ -99,6 +99,12 @@
<version>4.0.8.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-codec-http</artifactId>
<version>4.0.8.Final</version>
</dependency>
<dependency>
<groupId>org.fusesource.hawtbuf</groupId>
<artifactId>hawtbuf</artifactId>
@@ -139,6 +145,13 @@
<version>0.4.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-client</artifactId>
<version>9.2.0.M1</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java
+1
View File
@@ -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", "");
}
broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
+1 -1
View File
@@ -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
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
+104 -21
View File
@@ -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<BinaryWebSocketFrame> {
@Override
protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List<Object> out) throws Exception {
//convert the frame to a ByteBuf
ByteBuf bb = frame.content();
out.add(bb);
}
}
static class ByteBufToWebSocketFrameEncoder extends MessageToMessageEncoder<ByteBuf> {
@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);
}
}
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<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
b.group(m_bossGroup, m_workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@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<SocketChannel>() {
@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) {
broker/src/test/java/org/dna/mqtt/moquette/server/MQTTWebSocket.java
+83
View File
@@ -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);
}
}
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java
+1 -1
View File
@@ -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);
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationWebSocketTest.java
+66
View File
@@ -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);
}
}
distribution/src/main/resources/moquette.conf
+3 -1
View File
@@ -7,7 +7,9 @@
port 1883
websocket_port 8080
host 0.0.0.0
#Password file
password_file config/password_file.conf
password_file config/password_file.conf
parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java
+1
View File
@@ -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;
}