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

Added Netty connection stuff for the I/O

Browse Source
This commit is contained in:
andrea
2013-09-15 17:18:26 +02:00
parent 63842088a4
commit ac5fb3c9ac
12 changed files with 298 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
+6
View File
@@ -24,6 +24,12 @@
<artifactId>moquette-mina-parser</artifactId>
<version>0.2</version>
</dependency>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-netty-parser</artifactId>
<version>0.2</version>
</dependency>
<dependency>
<groupId>org.apache.mina</groupId>
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+2 -1
View File
@@ -89,7 +89,7 @@ class ProtocolProcessor {
int keepAlive = msg.getKeepAlive();
LOG.debug(String.format("Connect with keepAlive %d s", keepAlive));
session.setAttribute("keepAlive", keepAlive);
session.setAttribute(Constants.KEEP_ALIVE, keepAlive);
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
@@ -186,6 +186,7 @@ class ProtocolProcessor {
}
m_storageService.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
LOG.info("replying with PubAck to MSG ID " + evt.getMessageID());
}
if (retain) {
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+3 -1
View File
@@ -65,6 +65,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
SequenceBarrier barrier = m_ringBuffer.newBarrier();
m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
//TODO in a presentation is said to don't do the followinf line!!
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
m_executor.submit(m_eventProcessor);
@@ -73,6 +74,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private void disruptorPublish(MessagingEvent msgEvent) {
LOG.info("disruptorPublish publishing event " + msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
@@ -96,7 +98,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
LOG.debug("onEvent processing messaging event " + evt);
LOG.info("onEvent processing messaging event " + evt);
if (evt instanceof PublishEvent) {
m_processor.processPublish((PublishEvent) evt);
} else if (evt instanceof StopEvent) {
broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java
+1
View File
@@ -6,4 +6,5 @@ package org.dna.mqtt.moquette.server;
public class Constants {
public static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession";
public static final String KEEP_ALIVE = "keepAlive";
}
broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
+34 -3
View File
@@ -4,6 +4,7 @@ import java.io.File;
import java.io.IOException;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
import org.dna.mqtt.moquette.server.mina.MinaAcceptor;
import org.dna.mqtt.moquette.server.netty.NettyAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
@@ -16,13 +17,27 @@ public class Server {
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
File.separator + "moquette_store.hawtdb";
public static String NETTY_NIO_FRAMEWORK = "netty";
public static String MINA_NIO_FRAMEWORK = "mina";
public static String DEFAULT_NIO_FRAMEWORK = MINA_NIO_FRAMEWORK;
private ServerAcceptor m_acceptor;
SimpleMessaging messaging;
public static void main(String[] args) throws IOException {
String nioFrameworkType = DEFAULT_NIO_FRAMEWORK;
if (args.length > 0) {
//TODO check the typing and in case print usge message
nioFrameworkType = args[1];
if (!(nioFrameworkType.equals(DEFAULT_NIO_FRAMEWORK) ||
nioFrameworkType.equals(DEFAULT_NIO_FRAMEWORK))) {
printUsage(nioFrameworkType);
return;
}
}
final Server server = new Server();
server.startServer();
server.startServer(nioFrameworkType);
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
@@ -34,10 +49,19 @@ public class Server {
}
public void startServer() throws IOException {
startServer(DEFAULT_NIO_FRAMEWORK);
}
public void startServer(String nioFrameworkType) throws IOException {
LOG.info("Starting server with " + nioFrameworkType + " connectors");
messaging = SimpleMessaging.getInstance();
messaging.init();
m_acceptor = new MinaAcceptor();
if (nioFrameworkType.equals(MINA_NIO_FRAMEWORK)) {
m_acceptor = new MinaAcceptor();
} else if (nioFrameworkType.equals(NETTY_NIO_FRAMEWORK)) {
m_acceptor = new NettyAcceptor();
}
m_acceptor.initialize(messaging);
}
@@ -47,5 +71,12 @@ public class Server {
m_acceptor.close();
LOG.info("Server stopped");
}
private static void printUsage(String nioFrameworkType) {
System.out.println("The broker should be invoked with the following command line: \n");
System.out.println("> java -jar <broker.jar> [mina|netty]\n");
System.out.println("while was invoked with type: " + nioFrameworkType);
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java
+1 -2
View File
@@ -41,7 +41,6 @@ import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.server.MQTTHandler;
import org.dna.mqtt.moquette.server.ServerAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,7 +86,7 @@ public class MinaAcceptor implements ServerAcceptor {
m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") );
m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
MQTTHandler handler = new MQTTHandler();
MinaMQTTHandler handler = new MinaMQTTHandler();
handler.setMessaging(messaging);
broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java
-2
View File
@@ -35,6 +35,4 @@ public class MinaChannel implements ServerChannel {
public void setIdleTime(int idleTime) {
m_session.getConfig().setIdleTime(IdleStatus.READER_IDLE, idleTime);
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java → broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java
+11 -7
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.server;
package org.dna.mqtt.moquette.server.mina;
import io.netty.channel.ChannelHandler;
import java.util.HashMap;
import java.util.Map;
import org.apache.mina.core.service.IoHandlerAdapter;
@@ -10,7 +11,6 @@ import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.dna.mqtt.moquette.server.mina.MinaChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -19,9 +19,9 @@ import org.slf4j.LoggerFactory;
*
* @author andrea
*/
public class MQTTHandler extends IoHandlerAdapter {
public class MinaMQTTHandler extends IoHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
private static final Logger LOG = LoggerFactory.getLogger(MinaMQTTHandler.class);
private IMessaging m_messaging;
private Map<IoSession, MinaChannel> m_channelMapper = new HashMap<IoSession, MinaChannel>();
@@ -39,11 +39,15 @@ public class MQTTHandler extends IoHandlerAdapter {
case PUBCOMP:
case PUBREL:
case DISCONNECT:
if (!m_channelMapper.containsKey(session)) {
m_channelMapper.put(session, new MinaChannel(session));
MinaChannel channel;
synchronized(m_channelMapper) {
if (!m_channelMapper.containsKey(session)) {
m_channelMapper.put(session, new MinaChannel(session));
}
channel = m_channelMapper.get(session);
}
m_messaging.handleProtocolMessage(m_channelMapper.get(session), msg);
m_messaging.handleProtocolMessage(channel, msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
+79
View File
@@ -0,0 +1,79 @@
package org.dna.mqtt.moquette.server.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.io.IOException;
import org.dna.mqtt.commons.Constants;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.parser.netty.MQTTDecoder;
import org.dna.mqtt.moquette.parser.netty.MQTTEncoder;
import org.dna.mqtt.moquette.server.ServerAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class NettyAcceptor implements ServerAcceptor {
private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class);
EventLoopGroup m_bossGroup;
EventLoopGroup m_workerGroup;
public void initialize(IMessaging messaging) 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("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
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);
LOG.info("Server binded");
f.sync();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
public void close() {
if (m_workerGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
if (m_bossGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
m_workerGroup.shutdownGracefully();
m_bossGroup.shutdownGracefully();
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
+86
View File
@@ -0,0 +1,86 @@
package org.dna.mqtt.moquette.server.netty;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import java.util.HashMap;
import java.util.Map;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;
/**
*
* @author andrea
*/
public class NettyChannel implements ServerChannel {
private static class IdleTimoutHandler extends ChannelDuplexHandler {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleState) {
IdleState e = (IdleState) evt;
if (e == IdleState.ALL_IDLE) {
ctx.close();
} /*else if (e.getState() == IdleState.WRITER_IDLE) {
ctx.writeAndFlush(new PingMessage());
}*/
}
}
}
private ChannelHandlerContext m_channel;
private Map<Object, AttributeKey<Object>> m_attributesKeys = new HashMap<Object, AttributeKey<Object>>();
private static final AttributeKey<Object> ATTR_KEY_KEEPALIVE = new AttributeKey<Object>(Constants.KEEP_ALIVE);
private static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = new AttributeKey<Object>(Constants.CLEAN_SESSION);
private static final AttributeKey<Object> ATTR_KEY_CLIENTID = new AttributeKey<Object>(Constants.ATTR_CLIENTID);
NettyChannel(ChannelHandlerContext ctx) {
m_channel = ctx;
m_attributesKeys.put(Constants.KEEP_ALIVE, ATTR_KEY_KEEPALIVE);
m_attributesKeys.put(Constants.CLEAN_SESSION, ATTR_KEY_CLEANSESSION);
m_attributesKeys.put(Constants.ATTR_CLIENTID, ATTR_KEY_CLIENTID);
}
public Object getAttribute(Object key) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
return attr.get();
}
public void setAttribute(Object key, Object value) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
attr.set(value);
}
private synchronized AttributeKey<Object> mapKey(Object key) {
if (!m_attributesKeys.containsKey(key)) {
throw new IllegalArgumentException("mapKey can't find a matching AttributeKey for " + key);
}
return m_attributesKeys.get(key);
}
public void setIdleTime(int idleTime) {
if (m_channel.pipeline().names().contains("idleStateHandler")) {
m_channel.pipeline().remove("idleStateHandler");
}
if (m_channel.pipeline().names().contains("idleEventHandler")) {
m_channel.pipeline().remove("idleEventHandler");
}
m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new IdleTimoutHandler());
}
public void close(boolean immediately) {
m_channel.close();
}
public void write(Object value) {
m_channel.writeAndFlush(value);
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java
+72
View File
@@ -0,0 +1,72 @@
package org.dna.mqtt.moquette.server.netty;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.util.HashMap;
import java.util.Map;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
@Sharable
public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
private IMessaging m_messaging;
private Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
try {
switch (msg.getMessageType()) {
case CONNECT:
case SUBSCRIBE:
case UNSUBSCRIBE:
case PUBLISH:
case PUBREC:
case PUBCOMP:
case PUBREL:
case DISCONNECT:
NettyChannel channel;
synchronized(m_channelMapper) {
if (!m_channelMapper.containsKey(ctx)) {
m_channelMapper.put(ctx, new NettyChannel(ctx));
}
channel = m_channelMapper.get(ctx);
}
m_messaging.handleProtocolMessage(channel, msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
ctx.write(pingResp);
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx)throws Exception {
ctx.close(/*false*/);
synchronized(m_channelMapper) {
m_channelMapper.remove(ctx);
}
}
public void setMessaging(IMessaging messaging) {
m_messaging = messaging;
}
}
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java
+3 -3
View File
@@ -17,7 +17,7 @@ import static org.junit.Assert.*;
* Test the broker using the widely used FuseSource libraries
* */
public class ServerIntegrationFuseTest {
Server m_server;
MQTT m_mqtt;
BlockingConnection m_connection;
@@ -57,9 +57,9 @@ public class ServerIntegrationFuseTest {
Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
byte[] futSub = m_connection.subscribe(topics);
m_connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, false);
Message msg = m_connection.receive();
msg.ack();
assertEquals("/topic", msg.getTopic());