From beb556c73c911bd522813458e973982c090c6703 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 17 Jan 2016 11:11:36 +0100 Subject: [PATCH] First clean of Moquette's channel abstraction to user Netty's one --- .../moquette/server/ConnectionDescriptor.java | 6 +- .../server/netty/NettyMQTTHandler.java | 4 +- .../moquette/spi/impl/ProtocolProcessor.java | 56 +++++++++++-------- .../io/moquette/spi/impl/DummyChannel.java | 26 ++++----- .../spi/impl/ProtocolProcessorTest.java | 7 ++- broker/src/test/resources/log4j.properties | 11 ++-- 6 files changed, 60 insertions(+), 50 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 7be84ef1..3249e65e 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -15,6 +15,8 @@ */ package io.moquette.server; +import io.netty.channel.Channel; + /** * Value object to maintain the information of single connection, like ClientID, IoSession, * and other clean session fla. @@ -25,10 +27,10 @@ package io.moquette.server; public class ConnectionDescriptor { public final String clientID; - public final ServerChannel session; + public final Channel session; public final boolean cleanSession; - public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) { + public ConnectionDescriptor(String clientID, Channel session, boolean cleanSession) { this.clientID = clientID; this.session = session; this.cleanSession = cleanSession; diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 963d43d4..b1fa11a0 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -49,7 +49,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { try { switch (msg.getMessageType()) { case CONNECT: - m_processor.processConnect(new NettyChannel(ctx), (ConnectMessage) msg); + m_processor.processConnect(ctx.channel(), (ConnectMessage) msg); break; case SUBSCRIBE: m_processor.processSubscribe(new NettyChannel(ctx), (SubscribeMessage) msg); @@ -96,7 +96,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { if (stolenAttr != null && stolenAttr == Boolean.TRUE) { stolen = stolenAttr; } - m_processor.processConnectionLost(clientID, stolen, new NettyChannel(ctx)); + m_processor.processConnectionLost(clientID, stolen, ctx.channel()); } ctx.close(); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index e2b9500c..4a03d6d7 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -47,6 +47,9 @@ import io.moquette.proto.messages.SubAckMessage; import io.moquette.proto.messages.SubscribeMessage; import io.moquette.proto.messages.UnsubAckMessage; import io.moquette.proto.messages.UnsubscribeMessage; +import io.netty.channel.Channel; +import io.netty.channel.ChannelPipeline; +import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -133,14 +136,14 @@ public class ProtocolProcessor { m_sessionsStore = sessionsStore; } - public void processConnect(ServerChannel session, ConnectMessage msg) { + public void processConnect(Channel session, ConnectMessage msg) { LOG.debug("CONNECT for client <{}>", msg.getClientID()); if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); LOG.warn("processConnect sent bad proto ConnAck"); session.write(badProto); - session.close(false); + session.close(); return; } @@ -165,7 +168,7 @@ public class ProtocolProcessor { failedCredentials(session); return; } - session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername()); + session.attr(NettyChannel.ATTR_KEY_USERNAME).set(msg.getUsername()); } else if (!this.allowAnonymous) { failedCredentials(session); return; @@ -175,11 +178,11 @@ public class ProtocolProcessor { if (m_clientIDs.containsKey(msg.getClientID())) { LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); //clean the subscriptions if the old used a cleanSession = true - ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).session; + Channel oldSession = m_clientIDs.get(msg.getClientID()).session; ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID()); oldClientSession.disconnect(); - oldSession.setAttribute(NettyChannel.ATTR_KEY_SESSION_STOLEN, true); - oldSession.close(false); + oldSession.attr(NettyChannel.ATTR_KEY_SESSION_STOLEN).set(true); + oldSession.close(); LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); } @@ -188,13 +191,13 @@ public class ProtocolProcessor { int keepAlive = msg.getKeepAlive(); LOG.debug("Connect with keepAlive {} s", keepAlive); - session.setAttribute(NettyChannel.ATTR_KEY_KEEPALIVE, keepAlive); - session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, msg.isCleanSession()); + session.attr(NettyChannel.ATTR_KEY_KEEPALIVE).set(keepAlive); + session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); //used to track the client in the subscription and publishing phases. - session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, msg.getClientID()); + session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(msg.getClientID()); LOG.debug("Connect create session <{}>", session); - session.setIdleTime(Math.round(keepAlive * 1.5f)); + setIdleTime(session.pipeline(), Math.round(keepAlive * 1.5f)); //Handle will flag if (msg.isWillFlag()) { @@ -234,11 +237,17 @@ public class ProtocolProcessor { // LOG.info("CONNECT clients descriptors {}", m_clientIDs); } - private void failedCredentials(ServerChannel session) { + private void setIdleTime(ChannelPipeline pipeline, int idleTime) { + if (pipeline.names().contains("idleStateHandler")) { + pipeline.remove("idleStateHandler"); + } + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); + } + + private void failedCredentials(Channel session) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); session.write(okResp); - session.close(false); } /** @@ -289,12 +298,12 @@ public class ProtocolProcessor { return pub; } - public void processPublish(ServerChannel session, PublishMessage msg) { + public void processPublish(Channel session, PublishMessage msg) { LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); final String topic = msg.getTopicName(); //check if the topic can be wrote - String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); + String user = (String) session.attr(NettyChannel.ATTR_KEY_USERNAME).get(); if (!m_authorizator.canWrite(topic, user, clientID)) { LOG.debug("topic {} doesn't have write credentials", topic); return; @@ -518,10 +527,10 @@ public class ProtocolProcessor { // could happen is not an error HANDLE IT throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); } - ServerChannel session = m_clientIDs.get(clientId).session; + Channel session = m_clientIDs.get(clientId).session; LOG.debug("Session for clientId {} is {}", clientId, session); - String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); + String user = (String) session.attr(NettyChannel.ATTR_KEY_USERNAME).get(); if (!m_authorizator.canRead(topic, user, clientId)) { LOG.debug("topic {} doesn't have read credentials", topic); return; @@ -632,7 +641,7 @@ public class ProtocolProcessor { LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession); } - public void processConnectionLost(String clientID, boolean sessionStolen, NettyChannel channel) { + public void processConnectionLost(String clientID, boolean sessionStolen, Channel channel) { ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); m_clientIDs.remove(clientID, oldConnDescr); //If already removed a disconnect message was already processed for this clientID @@ -654,10 +663,11 @@ public class ProtocolProcessor { * Remove the clientID from topic subscription, if not previously subscribed, * doesn't reply any error */ - public void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) { + public void processUnsubscribe(Channel session, UnsubscribeMessage msg) { List topics = msg.topicFilters(); int messageID = msg.getMessageID(); - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); @@ -666,7 +676,7 @@ public class ProtocolProcessor { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { //close the connection, not valid topicFilter is a protocol violation - session.close(true); + session.close(); LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID); return; } @@ -684,8 +694,8 @@ public class ProtocolProcessor { session.write(ackMessage); } - public void processSubscribe(ServerChannel session, SubscribeMessage msg) { - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + public void processSubscribe(Channel session, SubscribeMessage msg) { + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); diff --git a/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java b/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java index f175e853..5090b05a 100644 --- a/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java +++ b/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java @@ -15,7 +15,7 @@ */ package io.moquette.spi.impl; -import io.moquette.server.ServerChannel; +import io.netty.channel.embedded.EmbeddedChannel; import io.netty.util.AttributeKey; import io.moquette.proto.messages.AbstractMessage; import io.moquette.proto.messages.ConnAckMessage; @@ -27,7 +27,7 @@ import java.util.Map; * * @author andrea */ -public class DummyChannel implements ServerChannel { +public class DummyChannel extends EmbeddedChannel { AbstractMessage m_receivedMessage; @@ -56,17 +56,17 @@ public class DummyChannel implements ServerChannel { return this.m_channelClosed; } - public void write(Object value) { - try { - m_receivedMessage = (AbstractMessage) value; - if (m_receivedMessage instanceof ConnAckMessage) { - ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; - m_returnCode = buf.getReturnCode(); - } - } catch (Exception ex) { - throw new AssertionError("Wrong return code"); - } - } +// public void write(Object value) { +// try { +// m_receivedMessage = (AbstractMessage) value; +// if (m_receivedMessage instanceof ConnAckMessage) { +// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; +// m_returnCode = buf.getReturnCode(); +// } +// } catch (Exception ex) { +// throw new AssertionError("Wrong return code"); +// } +// } public byte getReturnCode() { return m_returnCode; diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 659233f2..af5056c9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -28,6 +28,7 @@ import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.netty.channel.embedded.EmbeddedChannel; import org.junit.Before; import org.junit.Test; @@ -163,8 +164,8 @@ public class ProtocolProcessorTest { storageService.initStore(); subs.init(storageService.sessionsStore()); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); - - MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); + + EmbeddedChannel firstReceiverSession = new EmbeddedChannel(); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(FAKE_CLIENT_ID); @@ -172,7 +173,7 @@ public class ProtocolProcessorTest { m_processor.processConnect(firstReceiverSession, connectMessage); //connect the second fake subscriber - MockReceiverChannel secondReceiverSession = new MockReceiverChannel(); + EmbeddedChannel secondReceiverSession = new EmbeddedChannel(); ConnectMessage connectMessage2 = new ConnectMessage(); connectMessage2.setProtocolVersion((byte) 3); connectMessage2.setClientID(FAKE_CLIENT_ID2); diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 390ba3a9..2ce138c3 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -4,8 +4,8 @@ log4j.rootLogger=ERROR, stdout, file log4j.logger.io.moquette=WARN #log4j.logger.io.moquette=DEBUG -log4j.logger.io.moquette.spi.impl.ProtocolProcessor=TRACE -#log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN +#log4j.logger.io.moquette.spi.impl.ProtocolProcessor=TRACE +log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN #log4j.logger.io.moquette.spi.impl.SimpleMessaging=DEBUG #Protocol parsing @@ -17,15 +17,12 @@ log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN #log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG #log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=DEBUG log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN -log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=DEBUG +log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=WARN # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender -#log4j.appender.stdout.Threshold=WARN -#log4j.appender.stdout.Threshold=INFO -log4j.appender.stdout.Threshold=TRACE +log4j.appender.stdout.Threshold=INFO # for debug trace -#log4j.appender.stdout.Threshold=TRACE log4j.appender.stdout.layout=org.apache.log4j.PatternLayout #log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %M %L %x - %m%n