From bf3fcadd28365204b15602a24a5af1d363456380 Mon Sep 17 00:00:00 2001 From: andrea Date: Tue, 10 Mar 2015 23:06:37 +0100 Subject: [PATCH] Removed unused abstract key sessions (it's bound to Netty at the end, so hasn't sense --- .../moquette/server/ServerChannel.java | 6 ++- .../moquette/server/netty/NettyChannel.java | 29 +++++---------- .../server/netty/NettyMQTTHandler.java | 2 +- .../moquette/spi/impl/ProtocolProcessor.java | 30 ++++++++------- .../spi/impl/ProtocolProcessorTest.java | 37 ++++++++++--------- 5 files changed, 50 insertions(+), 54 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/server/ServerChannel.java b/broker/src/main/java/org/eclipse/moquette/server/ServerChannel.java index c4f099c2..1eeca9f7 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/ServerChannel.java +++ b/broker/src/main/java/org/eclipse/moquette/server/ServerChannel.java @@ -15,15 +15,17 @@ */ package org.eclipse.moquette.server; +import io.netty.util.AttributeKey; + /** * * @author andrea */ public interface ServerChannel { - Object getAttribute(Object key); + Object getAttribute(AttributeKey key); - void setAttribute(Object key, Object value); + void setAttribute(AttributeKey key, Object value); void setIdleTime(int idleTime); diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java index e89d0361..4bdd97b2 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java @@ -32,36 +32,25 @@ public class NettyChannel implements ServerChannel { private ChannelHandlerContext m_channel; - private Map> m_attributesKeys = new HashMap>(); - - 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); + + 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); 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)); + public Object getAttribute(AttributeKey key) { + Attribute attr = m_channel.attr(key); return attr.get(); } - public void setAttribute(Object key, Object value) { - Attribute attr = m_channel.attr(mapKey(key)); + public void setAttribute(AttributeKey key, Object value) { + Attribute attr = m_channel.attr(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"); @@ -83,7 +72,7 @@ public class NettyChannel implements ServerChannel { @Override public String toString() { - String clientID = (String) getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) getAttribute(ATTR_KEY_CLIENTID); return "session [clientID: "+ clientID +"]" + super.toString(); } } diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java index f4637933..f9e2d71c 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java @@ -78,7 +78,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyChannel channel = m_channelMapper.get(ctx); - String clientID = (String) channel.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); m_messaging.lostConnection(channel, clientID); ctx.close(/*false*/); synchronized(m_channelMapper) { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 19663a7e..ce01f816 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -22,6 +22,8 @@ import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; + +import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; @@ -160,7 +162,7 @@ class ProtocolProcessor implements EventHandler { 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()).getSession(); - boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); + boolean cleanSession = (Boolean) oldSession.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION); if (cleanSession) { //cleanup topic subscriptions cleanSession(msg.getClientID()); @@ -175,10 +177,10 @@ class ProtocolProcessor implements EventHandler { int keepAlive = msg.getKeepAlive(); LOG.debug("Connect with keepAlive {} s", keepAlive); - session.setAttribute(Constants.KEEP_ALIVE, keepAlive); - session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); + session.setAttribute(NettyChannel.ATTR_KEY_KEEPALIVE, keepAlive); + session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. - session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); + session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, msg.getClientID()); LOG.debug("Conent create session <{}>", session); session.setIdleTime(Math.round(keepAlive * 1.5f)); @@ -253,7 +255,7 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = PubAckMessage.class) void processPubAck(ServerChannel session, PubAckMessage msg) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); //Remove the message from message store m_messagesStore.removeMessageInSession(clientID, messageID); @@ -272,7 +274,7 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = PublishMessage.class) void processPublish(ServerChannel session, PublishMessage msg) { LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", msg); - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); final String topic = msg.getTopicName(); final AbstractMessage.QOSType qos = msg.getQos(); final ByteBuffer message = msg.getPayload(); @@ -449,7 +451,7 @@ class ProtocolProcessor implements EventHandler { * */ @MQTTMessage(message = PubRelMessage.class) void processPubRel(ServerChannel session, PubRelMessage msg) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); String publishKey = String.format("%s%d", clientID, messageID); @@ -480,7 +482,7 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = PubRecMessage.class) void processPubRec(ServerChannel session, PubRecMessage msg) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); //once received a PUBREC reply with a PUBREL(messageID) LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); @@ -495,7 +497,7 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = PubCompMessage.class) void processPubComp(ServerChannel session, PubCompMessage msg) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); 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 @@ -504,8 +506,8 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = DisconnectMessage.class) void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION); if (cleanSession) { //cleanup topic subscriptions cleanSession(clientID); @@ -554,7 +556,7 @@ class ProtocolProcessor implements EventHandler { void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) { List topics = msg.topicFilters(); int messageID = msg.getMessageID(); - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); for (String topic : topics) { @@ -570,8 +572,8 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = SubscribeMessage.class) void processSubscribe(ServerChannel session, SubscribeMessage msg) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); + String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION); LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); for (SubscribeMessage.Couple req : msg.subscriptions()) { diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index f066bea0..f49bf8c7 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -19,6 +19,9 @@ import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; + +import io.netty.util.AttributeKey; +import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; @@ -70,11 +73,11 @@ public class ProtocolProcessorTest { private Map m_attributes = new HashMap(); - public Object getAttribute(Object key) { + public Object getAttribute(AttributeKey key) { return m_attributes.get(key); } - public void setAttribute(Object key, Object value) { + public void setAttribute(AttributeKey key, Object value) { m_attributes.put(key, value); } @@ -106,11 +109,11 @@ public class ProtocolProcessorTest { AbstractMessage m_receivedMessage; private Map m_attributes = new HashMap(); - public Object getAttribute(Object key) { + public Object getAttribute(AttributeKey key) { return m_attributes.get(key); } - public void setAttribute(Object key, Object value) { + public void setAttribute(AttributeKey key, Object value) { m_attributes.put(key, value); } @@ -256,8 +259,8 @@ public class ProtocolProcessorTest { connMsg.setProcotolVersion(VERSION_3_1_1); connMsg.setClientID("CliID"); connMsg.setCleanSession(false); - m_session.setAttribute(Constants.ATTR_CLIENTID, "CliID"); - m_session.setAttribute(Constants.CLEAN_SESSION, false); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "CliID"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); //Connect a first time m_processor.processConnect(m_session, connMsg); @@ -310,7 +313,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.setAttribute(Constants.ATTR_CLIENTID, "FakeCLI"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "FakeCLI"); m_processor.processPublish(m_session, msg); //Verify @@ -364,7 +367,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.setAttribute(Constants.ATTR_CLIENTID, "FakeCLI"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "FakeCLI"); m_processor.processPublish(m_session, msg); //Verify @@ -385,8 +388,8 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_CLIENT_ID); - m_session.setAttribute(Constants.CLEAN_SESSION, false); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); //Verify @@ -399,8 +402,8 @@ public class ProtocolProcessorTest { public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_CLIENT_ID); - m_session.setAttribute(Constants.CLEAN_SESSION, false); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); subscriptions.clearAllSubscriptions(); assertEquals(0, subscriptions.size()); @@ -469,9 +472,9 @@ public class ProtocolProcessorTest { pubmsg.setQos(QOSType.MOST_ONE); pubmsg.setPayload(buffer); pubmsg.setRetainFlag(true); - m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_PUBLISHER_ID); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_PUBLISHER_ID); m_processor.processPublish(m_session, pubmsg); - m_session.setAttribute(Constants.CLEAN_SESSION, false); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); //Exercise SubscribeMessage msg = new SubscribeMessage(); @@ -524,7 +527,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "Publisher"); m_processor.processPublish(m_session, msg); //Verify no message is received @@ -549,7 +552,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "Publisher"); m_processor.processPublish(m_session, msg); //Verify no message is received @@ -567,7 +570,7 @@ public class ProtocolProcessorTest { connMsg.setClientID("Publisher"); m_processor.processConnect(m_session, connMsg); //prepare and existing retained store - m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "Publisher"); ByteBuffer payload = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage msg = new PublishMessage(); msg.setTopicName(FAKE_TOPIC);