From 3be2667844368c3e582f88d617ab4079bb33bc58 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 24 Jan 2016 19:10:23 +0100 Subject: [PATCH] Minor refact, moved Channel's attributesKey accessible by utils methods --- .../server/netty/NettyMQTTHandler.java | 6 +-- .../io/moquette/server/netty/NettyUtils.java | 46 ++++++++++++++++--- .../moquette/spi/impl/ProtocolProcessor.java | 34 +++++++------- .../spi/impl/ProtocolProcessorTest.java | 30 ++++++------ .../impl/ProtocolProcessor_CONNECT_Test.java | 8 ++-- 5 files changed, 80 insertions(+), 44 deletions(-) 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 45d394f6..05dd4112 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -87,14 +87,14 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - String clientID = (String) NettyUtils.getAttribute(ctx, NettyUtils.ATTR_KEY_CLIENTID); + String clientID = NettyUtils.clientID(ctx.channel()); if (clientID != null && !clientID.isEmpty()) { //if the channel was of a correctly connected client, inform messaging //else it was of a not completed CONNECT message or sessionStolen boolean stolen = false; - Boolean stolenAttr = (Boolean) NettyUtils.getAttribute(ctx, NettyUtils.ATTR_KEY_SESSION_STOLEN); + Boolean stolenAttr = NettyUtils.sessionStolen(ctx.channel()); if (stolenAttr != null && stolenAttr == Boolean.TRUE) { - stolen = stolenAttr; + stolen = true; } m_processor.processConnectionLost(clientID, stolen, ctx.channel()); } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java index 1cd45c39..cab996b1 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java @@ -16,8 +16,8 @@ package io.moquette.server.netty; import io.moquette.server.Constants; +import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; -import io.netty.handler.timeout.IdleStateHandler; import io.netty.util.Attribute; import io.netty.util.AttributeKey; @@ -31,16 +31,50 @@ public class NettyUtils { public static final String ATTR_USERNAME = "username"; public static final String ATTR_SESSION_STOLEN = "sessionStolen"; - public static final AttributeKey ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE); - public static final AttributeKey ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION); - public static final AttributeKey ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID); - public static final AttributeKey ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME); - public static final AttributeKey ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN); + private static final AttributeKey ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE); + private static final AttributeKey ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION); + private static final AttributeKey ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID); + private static final AttributeKey ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME); + private static final AttributeKey ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN); public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey key) { Attribute attr = ctx.channel().attr(key); return attr.get(); } + public static void keepAlive(Channel channel, int keepAlive) { + channel.attr(NettyUtils.ATTR_KEY_KEEPALIVE).set(keepAlive); + } + public static void cleanSession(Channel channel, boolean cleanSession) { + channel.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(cleanSession); + } + + public static boolean cleanSession(Channel channel) { + return (Boolean) channel.attr(NettyUtils.ATTR_KEY_CLEANSESSION).get(); + } + + public static void clientID(Channel channel, String clientID) { + channel.attr(NettyUtils.ATTR_KEY_CLIENTID).set(clientID); + } + + public static String clientID(Channel channel) { + return (String) channel.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + } + + public static void userName(Channel channel, String username) { + channel.attr(NettyUtils.ATTR_KEY_USERNAME).set(username); + } + + public static String userName(Channel channel) { + return (String) channel.attr(NettyUtils.ATTR_KEY_USERNAME).get(); + } + + public static void sessionStolen(Channel channel, boolean value) { + channel.attr(NettyUtils.ATTR_KEY_SESSION_STOLEN).set(value); + } + + public static Boolean sessionStolen(Channel channel) { + return (Boolean) channel.attr(NettyUtils.ATTR_KEY_SESSION_STOLEN).get(); + } } 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 c5fb483d..a2b89fd0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -168,7 +168,7 @@ public class ProtocolProcessor { session.close(); return; } - session.attr(NettyUtils.ATTR_KEY_USERNAME).set(msg.getUsername()); + NettyUtils.userName(session, msg.getUsername()); } else if (!this.allowAnonymous) { failedCredentials(session); return; @@ -181,7 +181,7 @@ public class ProtocolProcessor { Channel oldSession = m_clientIDs.get(msg.getClientID()).session; ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID()); oldClientSession.disconnect(); - oldSession.attr(NettyUtils.ATTR_KEY_SESSION_STOLEN).set(true); + NettyUtils.sessionStolen(oldSession, true); oldSession.close(); LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); } @@ -191,10 +191,12 @@ public class ProtocolProcessor { int keepAlive = msg.getKeepAlive(); LOG.debug("Connect with keepAlive {} s", keepAlive); - session.attr(NettyUtils.ATTR_KEY_KEEPALIVE).set(keepAlive); - session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); + NettyUtils.keepAlive(session, keepAlive); + //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); + NettyUtils.cleanSession(session, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. - session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); + //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); + NettyUtils.clientID(session, msg.getClientID()); LOG.debug("Connect create session <{}>", session); setIdleTime(session.pipeline(), Math.round(keepAlive * 1.5f)); @@ -271,7 +273,7 @@ public class ProtocolProcessor { } public void processPubAck(Channel session, PubAckMessage msg) { - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); int messageID = msg.getMessageID(); //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -300,10 +302,10 @@ public class ProtocolProcessor { public void processPublish(Channel session, PublishMessage msg) { LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); final String topic = msg.getTopicName(); //check if the topic can be wrote - String user = (String) session.attr(NettyUtils.ATTR_KEY_USERNAME).get(); + String user = NettyUtils.userName(session); if (!m_authorizator.canWrite(topic, user, clientID)) { LOG.debug("topic {} doesn't have write credentials", topic); return; @@ -486,7 +488,7 @@ public class ProtocolProcessor { Channel session = m_clientIDs.get(clientId).session; LOG.debug("Session for clientId {} is {}", clientId, session); - String user = (String) session.attr(NettyUtils.ATTR_KEY_USERNAME).get(); + String user = NettyUtils.userName(session); if (!m_authorizator.canRead(topic, user, clientId)) { LOG.debug("topic {} doesn't have read credentials", topic); return; @@ -525,7 +527,7 @@ public class ProtocolProcessor { * to all interested subscribers. * */ public void processPubRel(Channel session, PubRelMessage msg) { - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -554,7 +556,7 @@ public class ProtocolProcessor { } public void processPubRec(Channel session, PubRecMessage msg) { - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); int messageID = msg.getMessageID(); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); @@ -571,7 +573,7 @@ public class ProtocolProcessor { } public void processPubComp(Channel session, PubCompMessage msg) { - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); int messageID = msg.getMessageID(); LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory @@ -581,8 +583,8 @@ public class ProtocolProcessor { } public void processDisconnect(Channel session) throws InterruptedException { - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); - boolean cleanSession = (Boolean) session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).get(); + String clientID = NettyUtils.clientID(session); + boolean cleanSession = NettyUtils.cleanSession(session); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); clientSession.disconnect(); @@ -622,7 +624,7 @@ public class ProtocolProcessor { public void processUnsubscribe(Channel session, UnsubscribeMessage msg) { List topics = msg.topicFilters(); int messageID = msg.getMessageID(); - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); @@ -651,7 +653,7 @@ public class ProtocolProcessor { } public void processSubscribe(Channel session, SubscribeMessage msg) { - String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + String clientID = NettyUtils.clientID(session); 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/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index d19adf46..4bbdc327 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -130,7 +130,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.attr(NettyUtils.ATTR_KEY_USERNAME).set("FakeCLI"); + NettyUtils.userName(m_session, "FakeCLI"); m_processor.processPublish(m_session, msg); //Verify @@ -189,7 +189,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.attr(NettyUtils.ATTR_KEY_USERNAME).set("FakeCLI"); + NettyUtils.userName(m_session, "FakeCLI"); m_processor.processPublish(m_session, msg); @@ -210,8 +210,8 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(false); + NettyUtils.clientID(m_session, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_session, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_session, msg); @@ -225,8 +225,8 @@ public class ProtocolProcessorTest { public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(false); + NettyUtils.clientID(m_session, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_session, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -246,8 +246,8 @@ public class ProtocolProcessorTest { public void testSubscribeWithBadFormattedTopic() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(false); + NettyUtils.clientID(m_session, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_session, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -271,8 +271,8 @@ public class ProtocolProcessorTest { UnsubscribeMessage msg = new UnsubscribeMessage(); msg.setMessageID(1); msg.addTopicFilter(BAD_FORMATTED_TOPIC); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(false); + NettyUtils.clientID(m_session, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_session, false); //Exercise m_processor.processUnsubscribe(m_session, msg); @@ -317,9 +317,9 @@ public class ProtocolProcessorTest { pubmsg.setQos(QOSType.MOST_ONE); pubmsg.setPayload(buffer); pubmsg.setRetainFlag(true); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_PUBLISHER_ID); + NettyUtils.clientID(m_session, FAKE_PUBLISHER_ID); m_processor.processPublish(m_session, pubmsg); - m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(false); + NettyUtils.cleanSession(m_session, false); //Exercise SubscribeMessage msg = new SubscribeMessage(); @@ -377,7 +377,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("Publisher"); + NettyUtils.clientID(m_session, "Publisher"); m_processor.processPublish(m_session, msg); //Verify no message is received @@ -404,7 +404,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("Publisher"); + NettyUtils.clientID(m_session, "Publisher"); m_processor.processPublish(m_session, msg); //Verify no message is received @@ -422,7 +422,7 @@ public class ProtocolProcessorTest { connMsg.setClientID("Publisher"); m_processor.processConnect(m_session, connMsg); //prepare and existing retained store - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("Publisher"); + NettyUtils.clientID(m_session, "Publisher"); ByteBuffer payload = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage msg = new PublishMessage(); msg.setTopicName(FAKE_TOPIC); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index c1cc1e34..830089a4 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -246,8 +246,8 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setProtocolVersion(VERSION_3_1_1); connMsg.setClientID("CliID"); connMsg.setCleanSession(false); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("CliID"); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(false); + NettyUtils.clientID(m_session, "CliID"); + NettyUtils.cleanSession(m_session, false); //Connect a first time m_processor.processConnect(m_session, connMsg); @@ -277,8 +277,8 @@ public class ProtocolProcessor_CONNECT_Test { SubscribeMessage subscribeMsg = new SubscribeMessage(); subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), ProtocolProcessorTest.FAKE_TOPIC)); - m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("CliID"); - m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(false); + NettyUtils.clientID(m_session, "CliID"); + NettyUtils.cleanSession(m_session, false); m_processor.processSubscribe(m_session, subscribeMsg); Subscription expectedSubscription = new Subscription("CliID", ProtocolProcessorTest.FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE);