From ac5fb3c9ac7bfbd48ecc4e3719457152c4109945 Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 15 Sep 2013 17:18:26 +0200 Subject: [PATCH] Added Netty connection stuff for the I/O --- broker/pom.xml | 6 ++ .../messaging/spi/impl/ProtocolProcessor.java | 3 +- .../messaging/spi/impl/SimpleMessaging.java | 4 +- .../dna/mqtt/moquette/server/Constants.java | 1 + .../org/dna/mqtt/moquette/server/Server.java | 37 +++++++- .../moquette/server/mina/MinaAcceptor.java | 3 +- .../moquette/server/mina/MinaChannel.java | 2 - .../MinaMQTTHandler.java} | 18 ++-- .../moquette/server/netty/NettyAcceptor.java | 79 +++++++++++++++++ .../moquette/server/netty/NettyChannel.java | 86 +++++++++++++++++++ .../server/netty/NettyMQTTHandler.java | 72 ++++++++++++++++ .../server/ServerIntegrationFuseTest.java | 6 +- 12 files changed, 298 insertions(+), 19 deletions(-) rename broker/src/main/java/org/dna/mqtt/moquette/server/{MQTTHandler.java => mina/MinaMQTTHandler.java} (75%) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java diff --git a/broker/pom.xml b/broker/pom.xml index e14f8dca..59744733 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -24,6 +24,12 @@ moquette-mina-parser 0.2 + + + org.dna.mqtt + moquette-netty-parser + 0.2 + org.apache.mina diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 1d457c3b..dd4053a2 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -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) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 0dd55221..7d42c95c 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -65,6 +65,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { SequenceBarrier barrier = m_ringBuffer.newBarrier(); m_eventProcessor = new BatchEventProcessor(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 { 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 { 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) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java index 0a2ea7fd..17a691c5 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java @@ -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"; } 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 d4872f78..5a9412bb 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 @@ -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 [mina|netty]\n"); + System.out.println("while was invoked with type: " + nioFrameworkType); + } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java index 324586aa..98be370b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java @@ -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); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java index be591d51..14064ddd 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java @@ -35,6 +35,4 @@ public class MinaChannel implements ServerChannel { public void setIdleTime(int idleTime) { m_session.getConfig().setIdleTime(IdleStatus.READER_IDLE, idleTime); } - - } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java similarity index 75% rename from broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java rename to broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java index 36cfdfe5..bc6d8ad9 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java @@ -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 m_channelMapper = new HashMap(); @@ -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(); 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 new file mode 100644 index 00000000..b10e576d --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -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() { + @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(); + } + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java new file mode 100644 index 00000000..e83aa609 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java @@ -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> m_attributesKeys = new HashMap>(); + + private static final AttributeKey ATTR_KEY_KEEPALIVE = new AttributeKey(Constants.KEEP_ALIVE); + private static final AttributeKey ATTR_KEY_CLEANSESSION = new AttributeKey(Constants.CLEAN_SESSION); + private static final AttributeKey ATTR_KEY_CLIENTID = new AttributeKey(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 attr = m_channel.attr(mapKey(key)); + return attr.get(); + } + + public void setAttribute(Object key, Object value) { + Attribute attr = m_channel.attr(mapKey(key)); + attr.set(value); + } + + private synchronized AttributeKey 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); + } + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java new file mode 100644 index 00000000..7e39e1ca --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java @@ -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 m_channelMapper = new HashMap(); + + @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; + } +} 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 cea7268e..11409151 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 @@ -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());