anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Minor refact, moved commons Channel attributes constants from NettyChannel to NettyUtils

Browse Source
This commit is contained in:
andsel
2016-01-24 19:21:24 +01:00
parent aa94d47148
commit 43371d3068
6 changed files with 58 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/NettyChannel.java
-91
View File
@@ -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<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
public static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
public static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
public static final AttributeKey<Object> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
public static final AttributeKey<Object> ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN);
NettyChannel(ChannelHandlerContext ctx) {
m_channel = ctx;
}
public Object getAttribute(AttributeKey<Object> key) {
Attribute<Object> attr = m_channel.attr(key);
return attr.get();
}
public void setAttribute(AttributeKey<Object> key, Object value) {
Attribute<Object> 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;
}
}
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+2 -2
View File
@@ -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;
}
broker/src/main/java/io/moquette/server/netty/NettyUtils.java
+10 -14
View File
@@ -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<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
public static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
public static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
public static final AttributeKey<Object> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
public static final AttributeKey<Object> ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN);
public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey<Object> key) {
Attribute<Object> attr = ctx.channel().attr(key);
return attr.get();
}
public static void setAttribute(ChannelHandlerContext ctx, AttributeKey<Object> key, Object value) {
Attribute<Object> 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());
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+17 -17
View File
@@ -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<String> 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);
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+16 -16
View File
@@ -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);
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+13 -9
View File
@@ -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);