diff --git a/broker/src/main/java/io/moquette/server/netty/NettyChannel.java b/broker/src/main/java/io/moquette/server/netty/NettyChannel.java deleted file mode 100644 index 222eaf33..00000000 --- a/broker/src/main/java/io/moquette/server/netty/NettyChannel.java +++ /dev/null @@ -1,91 +0,0 @@ -/* - * Copyright (c) 2012-2015 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package io.moquette.server.netty; - -import io.moquette.server.Constants; -import io.netty.channel.ChannelHandlerContext; -import io.netty.handler.timeout.IdleStateHandler; -import io.netty.util.Attribute; -import io.netty.util.AttributeKey; - -/** - * - * @author andrea - */ -public class NettyChannel { - - private ChannelHandlerContext m_channel; - - 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); - - NettyChannel(ChannelHandlerContext ctx) { - m_channel = ctx; - } - - public Object getAttribute(AttributeKey key) { - Attribute attr = m_channel.attr(key); - return attr.get(); - } - - public void setAttribute(AttributeKey key, Object value) { - Attribute attr = m_channel.attr(key); - attr.set(value); - } - - public void setIdleTime(int idleTime) { - if (m_channel.pipeline().names().contains("idleStateHandler")) { - m_channel.pipeline().remove("idleStateHandler"); - } - m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); - } - - public void close(boolean immediately) { - m_channel.close(); - } - - public void write(Object value) { - m_channel.writeAndFlush(value); - } - - @Override - public String toString() { - String clientID = (String) getAttribute(ATTR_KEY_CLIENTID); - return "session [clientID: "+ clientID +"]"; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - NettyChannel that = (NettyChannel) o; - - return !(m_channel != null ? !m_channel.channel().equals(that.m_channel.channel()) : that.m_channel != null); - - } - - @Override - public int hashCode() { - return m_channel != null ? m_channel.channel().hashCode() : 0; - } -} 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 cc9db83e..45d394f6 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -87,12 +87,12 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID); + String clientID = (String) NettyUtils.getAttribute(ctx, NettyUtils.ATTR_KEY_CLIENTID); 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, NettyChannel.ATTR_KEY_SESSION_STOLEN); + Boolean stolenAttr = (Boolean) NettyUtils.getAttribute(ctx, NettyUtils.ATTR_KEY_SESSION_STOLEN); if (stolenAttr != null && stolenAttr == Boolean.TRUE) { stolen = stolenAttr; } 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 2611eb3c..1cd45c39 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java @@ -15,6 +15,7 @@ */ package io.moquette.server.netty; +import io.moquette.server.Constants; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.timeout.IdleStateHandler; import io.netty.util.Attribute; @@ -27,24 +28,19 @@ import io.netty.util.AttributeKey; */ 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); + public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey key) { Attribute attr = ctx.channel().attr(key); return attr.get(); } - public static void setAttribute(ChannelHandlerContext ctx, AttributeKey key, Object value) { - Attribute attr = ctx.channel().attr(key); - attr.set(value); - } - public static void setIdleTime(ChannelHandlerContext channel, int idleTime) { - if (channel.pipeline().names().contains("idleStateHandler")) { - channel.pipeline().remove("idleStateHandler"); - } - if (channel.pipeline().names().contains("idleEventHandler")) { - channel.pipeline().remove("idleEventHandler"); - } - channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); - channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler()); - } } 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 d58f6958..c5fb483d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -21,7 +21,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import io.moquette.server.ConnectionDescriptor; -import io.moquette.server.netty.NettyChannel; +import io.moquette.server.netty.NettyUtils; import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; @@ -168,7 +168,7 @@ public class ProtocolProcessor { session.close(); return; } - session.attr(NettyChannel.ATTR_KEY_USERNAME).set(msg.getUsername()); + session.attr(NettyUtils.ATTR_KEY_USERNAME).set(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(NettyChannel.ATTR_KEY_SESSION_STOLEN).set(true); + oldSession.attr(NettyUtils.ATTR_KEY_SESSION_STOLEN).set(true); oldSession.close(); LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); } @@ -191,10 +191,10 @@ public class ProtocolProcessor { int keepAlive = msg.getKeepAlive(); LOG.debug("Connect with keepAlive {} s", keepAlive); - session.attr(NettyChannel.ATTR_KEY_KEEPALIVE).set(keepAlive); - session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); + session.attr(NettyUtils.ATTR_KEY_KEEPALIVE).set(keepAlive); + session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); //used to track the client in the subscription and publishing phases. - session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(msg.getClientID()); + session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); LOG.debug("Connect create session <{}>", session); setIdleTime(session.pipeline(), Math.round(keepAlive * 1.5f)); @@ -271,7 +271,7 @@ public class ProtocolProcessor { } public void processPubAck(Channel session, PubAckMessage msg) { - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); int messageID = msg.getMessageID(); //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -300,10 +300,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(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); final String topic = msg.getTopicName(); //check if the topic can be wrote - String user = (String) session.attr(NettyChannel.ATTR_KEY_USERNAME).get(); + String user = (String) session.attr(NettyUtils.ATTR_KEY_USERNAME).get(); if (!m_authorizator.canWrite(topic, user, clientID)) { LOG.debug("topic {} doesn't have write credentials", topic); return; @@ -486,7 +486,7 @@ public class ProtocolProcessor { Channel session = m_clientIDs.get(clientId).session; LOG.debug("Session for clientId {} is {}", clientId, session); - String user = (String) session.attr(NettyChannel.ATTR_KEY_USERNAME).get(); + String user = (String) session.attr(NettyUtils.ATTR_KEY_USERNAME).get(); if (!m_authorizator.canRead(topic, user, clientId)) { LOG.debug("topic {} doesn't have read credentials", topic); return; @@ -525,7 +525,7 @@ public class ProtocolProcessor { * to all interested subscribers. * */ public void processPubRel(Channel session, PubRelMessage msg) { - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); 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 +554,7 @@ public class ProtocolProcessor { } public void processPubRec(Channel session, PubRecMessage msg) { - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); int messageID = msg.getMessageID(); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); @@ -571,7 +571,7 @@ public class ProtocolProcessor { } public void processPubComp(Channel session, PubCompMessage msg) { - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); 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 +581,8 @@ public class ProtocolProcessor { } public void processDisconnect(Channel session) throws InterruptedException { - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); - boolean cleanSession = (Boolean) session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); + boolean cleanSession = (Boolean) session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).get(); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); clientSession.disconnect(); @@ -622,7 +622,7 @@ public class ProtocolProcessor { public void processUnsubscribe(Channel session, UnsubscribeMessage msg) { List topics = msg.topicFilters(); int messageID = msg.getMessageID(); - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.ATTR_KEY_CLIENTID).get(); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); @@ -651,7 +651,7 @@ public class ProtocolProcessor { } public void processSubscribe(Channel session, SubscribeMessage msg) { - String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + String clientID = (String) session.attr(NettyUtils.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/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 84bc3d3c..d19adf46 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -19,7 +19,7 @@ import io.moquette.proto.messages.*; import io.moquette.interception.InterceptHandler; import io.moquette.proto.messages.AbstractMessage.QOSType; import io.moquette.server.ConnectionDescriptor; -import io.moquette.server.netty.NettyChannel; +import io.moquette.server.netty.NettyUtils; import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; @@ -130,7 +130,7 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.attr(NettyChannel.ATTR_KEY_USERNAME).set("FakeCLI"); + m_session.attr(NettyUtils.ATTR_KEY_USERNAME).set("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(NettyChannel.ATTR_KEY_USERNAME).set("FakeCLI"); + m_session.attr(NettyUtils.ATTR_KEY_USERNAME).set("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(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(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(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(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(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(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(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); - m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(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(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_PUBLISHER_ID); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(FAKE_PUBLISHER_ID); m_processor.processPublish(m_session, pubmsg); - m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + m_session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(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(NettyChannel.ATTR_KEY_CLIENTID).set("Publisher"); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("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(NettyChannel.ATTR_KEY_CLIENTID).set("Publisher"); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("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(NettyChannel.ATTR_KEY_CLIENTID).set("Publisher"); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("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 04486a69..c1cc1e34 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 @@ -18,10 +18,10 @@ package io.moquette.spi.impl; import io.moquette.proto.messages.AbstractMessage; import io.moquette.proto.messages.ConnectMessage; import io.moquette.proto.messages.SubscribeMessage; +import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; -import io.moquette.server.netty.NettyChannel; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; import io.netty.channel.embedded.EmbeddedChannel; @@ -169,7 +169,8 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void prohibitAnonymousClient() { connMsg.setClientID("123"); - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, + new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -183,7 +184,8 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setClientID("123"); connMsg.setUserFlag(true); connMsg.setUsername(ProtocolProcessorTest.TEST_USER + "_fake"); - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, + new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -195,7 +197,8 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void acceptAnonymousClient() { connMsg.setClientID("123"); - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, + new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -243,8 +246,8 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setProtocolVersion(VERSION_3_1_1); connMsg.setClientID("CliID"); connMsg.setCleanSession(false); - m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("CliID"); - m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(false); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set("CliID"); + m_session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(false); //Connect a first time m_processor.processConnect(m_session, connMsg); @@ -272,9 +275,10 @@ public class ProtocolProcessor_CONNECT_Test { //subscribe SubscribeMessage subscribeMsg = new SubscribeMessage(); - subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), ProtocolProcessorTest.FAKE_TOPIC)); - m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("CliID"); - m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + 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); m_processor.processSubscribe(m_session, subscribeMsg); Subscription expectedSubscription = new Subscription("CliID", ProtocolProcessorTest.FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE);