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

Extracted common Netty bootstrapping method

Browse Source
This commit is contained in:
selva.andre
2014-08-26 21:55:42 +02:00
parent 2be0fd84fe
commit 7314dbee73
4 changed files with 89 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
+2 -2
View File
@@ -161,8 +161,8 @@
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId> <artifactId>maven-compiler-plugin</artifactId>
<configuration> <configuration>
<source>1.5</source> <source>1.6</source>
<target>1.5</target> <target>1.6</target>
</configuration> </configuration>
</plugin> </plugin>
broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
+1 -3
View File
@@ -38,7 +38,6 @@ public class Server {
SimpleMessaging messaging; SimpleMessaging messaging;
public static void main(String[] args) throws IOException { public static void main(String[] args) throws IOException {
final Server server = new Server(); final Server server = new Server();
server.startServer(); server.startServer();
System.out.println("Server started, version 0.7-SNAPSHOT"); System.out.println("Server started, version 0.7-SNAPSHOT");
@@ -49,7 +48,6 @@ public class Server {
server.stopServer(); server.stopServer();
} }
}); });
} }
/** /**
@@ -69,7 +67,7 @@ public class Server {
try { try {
confParser.parse(configFile); confParser.parse(configFile);
} catch (ParseException pex) { } catch (ParseException pex) {
LOG.warn("An error occured in parsing configuration, fallback on default configuration", pex); LOG.warn("An error occurred in parsing configuration, fallback on default configuration", pex);
} }
Properties configProps = confParser.getProperties(); Properties configProps = confParser.getProperties();
startServer(configProps); startServer(configProps);
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
+69 -94
View File
@@ -80,19 +80,21 @@ public class NettyAcceptor implements ServerAcceptor {
} }
} }
abstract class PipelineInitializer {
abstract void init(ChannelPipeline pipeline) throws Exception;
}
private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class); private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class);
EventLoopGroup m_bossGroup; EventLoopGroup m_bossGroup;
EventLoopGroup m_workerGroup; EventLoopGroup m_workerGroup;
//BytesMetricsCollector m_metricsCollector = new BytesMetricsCollector(); //BytesMetricsCollector m_metricsCollector = new BytesMetricsCollector();
MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector(); MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
MQTTEncoder mqttEncoder = new MQTTEncoder();
MQTTDecoder mqttDecoder = new MQTTDecoder();
@Override @Override
public void initialize(IMessaging messaging, Properties props) throws IOException { public void initialize(IMessaging messaging, Properties props) throws IOException {
m_bossGroup = new NioEventLoopGroup(); m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup(); m_workerGroup = new NioEventLoopGroup();
initializePlainTCPTransport(messaging, props); initializePlainTCPTransport(messaging, props);
@@ -100,13 +102,7 @@ public class NettyAcceptor implements ServerAcceptor {
initializeSSLTCPTransport(messaging, props); initializeSSLTCPTransport(messaging, props);
} }
private void initializePlainTCPTransport(IMessaging messaging, Properties props) throws IOException { private void initFactory(String host, int port, final PipelineInitializer pipeliner) {
// m_bossGroup = new NioEventLoopGroup();
// m_workerGroup = new NioEventLoopGroup();
final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging);
ServerBootstrap b = new ServerBootstrap(); ServerBootstrap b = new ServerBootstrap();
b.group(m_bossGroup, m_workerGroup) b.group(m_bossGroup, m_workerGroup)
.channel(NioServerSocketChannel.class) .channel(NioServerSocketChannel.class)
@@ -114,14 +110,7 @@ public class NettyAcceptor implements ServerAcceptor {
@Override @Override
public void initChannel(SocketChannel ch) throws Exception { public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline(); ChannelPipeline pipeline = ch.pipeline();
//pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector)); pipeliner.init(pipeline);
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_BACKLOG, 128)
@@ -129,8 +118,7 @@ public class NettyAcceptor implements ServerAcceptor {
.childOption(ChannelOption.SO_KEEPALIVE, true); .childOption(ChannelOption.SO_KEEPALIVE, true);
try { try {
// Bind and start to accept incoming connections. // Bind and start to accept incoming connections.
// ChannelFuture f = b.bind(Constants.PORT); ChannelFuture f = b.bind(host, port);
ChannelFuture f = b.bind(props.getProperty("host"), Integer.parseInt(props.getProperty("port")));
LOG.info("Server binded"); LOG.info("Server binded");
f.sync(); f.sync();
} catch (InterruptedException ex) { } catch (InterruptedException ex) {
@@ -138,6 +126,25 @@ public class NettyAcceptor implements ServerAcceptor {
} }
} }
private void initializePlainTCPTransport(IMessaging messaging, Properties props) throws IOException {
final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging);
String host = props.getProperty("host");
int port = Integer.parseInt(props.getProperty("port"));
initFactory(host, port, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) {
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);
}
});
}
private void initializeWebSocketTransport(IMessaging messaging, Properties props) throws IOException { private void initializeWebSocketTransport(IMessaging messaging, Properties props) throws IOException {
String webSocketPortProp = props.getProperty("websocket_port"); String webSocketPortProp = props.getProperty("websocket_port");
if (webSocketPortProp == null) { if (webSocketPortProp == null) {
@@ -150,38 +157,22 @@ public class NettyAcceptor implements ServerAcceptor {
final NettyMQTTHandler handler = new NettyMQTTHandler(); final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging); handler.setMessaging(messaging);
ServerBootstrap b = new ServerBootstrap(); String host = props.getProperty("host");
b.group(m_bossGroup, m_workerGroup) initFactory(host, port, new PipelineInitializer() {
.channel(NioServerSocketChannel.class) @Override
.childHandler(new ChannelInitializer<SocketChannel>() { void init(ChannelPipeline pipeline) {
@Override pipeline.addLast("httpEncoder", new HttpResponseEncoder());
public void initChannel(SocketChannel ch) throws Exception { pipeline.addLast("httpDdecoder", new HttpRequestDecoder());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqttv3.1"));
pipeline.addLast("bytebuf2wsDecoderr", new ByteBufToWebSocketFrameEncoder());
pipeline.addLast("httpEncoder", new HttpResponseEncoder()); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("httpDdecoder", new HttpRequestDecoder()); pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqttv3.1")); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("bytebuf2wsDecoderr", new ByteBufToWebSocketFrameEncoder()); pipeline.addLast("handler", handler);
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 {
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);
}
} }
@@ -215,52 +206,36 @@ public class NettyAcceptor implements ServerAcceptor {
} }
int sslPort = Integer.parseInt(sslPortProp); int sslPort = Integer.parseInt(sslPortProp);
String host = props.getProperty("host");
final NettyMQTTHandler handler = new NettyMQTTHandler(); final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging); handler.setMessaging(messaging);
initFactory(host, sslPort, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) throws Exception {
InputStream jksInputStream = getClass().getClassLoader().getResourceAsStream(jksPath);
SSLContext serverContext = SSLContext.getInstance("TLS");
final KeyStore ks = KeyStore.getInstance("JKS");
ks.load(jksInputStream, keyStorePassword.toCharArray());
final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(ks, keyManagerPassword.toCharArray());
serverContext.init(kmf.getKeyManagers(), null, null);
ServerBootstrap b = new ServerBootstrap(); SSLEngine engine = serverContext.createSSLEngine();
b.group(m_bossGroup, m_workerGroup) engine.setUseClientMode(false);
.channel(NioServerSocketChannel.class) final SslHandler sslHandler = new SslHandler(engine);
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
InputStream jksInputStream = getClass().getClassLoader().getResourceAsStream(jksPath);
SSLContext serverContext = SSLContext.getInstance("TLS");
final KeyStore ks = KeyStore.getInstance("JKS");
ks.load(jksInputStream, keyStorePassword.toCharArray());
final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(ks, keyManagerPassword.toCharArray());
serverContext.init(kmf.getKeyManagers(), null, null);
SSLEngine engine = serverContext.createSSLEngine(); pipeline.addLast("ssl", sslHandler);
engine.setUseClientMode(false); //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector));
final SslHandler sslHandler = new SslHandler(engine); pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
ChannelPipeline pipeline = ch.pipeline(); //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addLast("ssl", sslHandler); pipeline.addLast("decoder", new MQTTDecoder());
//pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector)); pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); pipeline.addLast("handler", handler);
//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"), sslPort);
LOG.info("Server binded");
f.sync();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
} }
public void close() { public void close() {
pom.xml
+13
View File
@@ -33,6 +33,19 @@
</plugins> </plugins>
</reporting> </reporting>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
</configuration>
</plugin>
</plugins>
</build>
<!--build> <!--build>
<pluginManagement> <pluginManagement>
<plugins> <plugins>