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

Removed unused abstract key sessions (it's bound to Netty at the end, so hasn't sense

Browse Source
This commit is contained in:
andrea
2015-03-10 23:06:37 +01:00
parent 715937a394
commit bf3fcadd28
5 changed files with 50 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/server/ServerChannel.java
+4 -2
View File
@@ -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<Object> key);
void setAttribute(Object key, Object value);
void setAttribute(AttributeKey<Object> key, Object value);
void setIdleTime(int idleTime);
broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java
+9 -20
View File
@@ -32,36 +32,25 @@ public class NettyChannel implements ServerChannel {
private ChannelHandlerContext m_channel;
private Map<Object, AttributeKey<Object>> m_attributesKeys = new HashMap<Object, AttributeKey<Object>>();
private static final AttributeKey<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
private static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
private static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
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);
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<Object> attr = m_channel.attr(mapKey(key));
public Object getAttribute(AttributeKey<Object> key) {
Attribute<Object> attr = m_channel.attr(key);
return attr.get();
}
public void setAttribute(Object key, Object value) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
public void setAttribute(AttributeKey<Object> key, Object value) {
Attribute<Object> attr = m_channel.attr(key);
attr.set(value);
}
private synchronized AttributeKey<Object> 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();
}
}
broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java
+1 -1
View File
@@ -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) {
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+16 -14
View File
@@ -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<ValueEvent> {
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<ValueEvent> {
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<ValueEvent> {
@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<ValueEvent> {
@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<ValueEvent> {
* */
@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<ValueEvent> {
@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<ValueEvent> {
@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<ValueEvent> {
@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<ValueEvent> {
void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
List<String> 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<ValueEvent> {
@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()) {
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java
+20 -17
View File
@@ -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<Object, Object> m_attributes = new HashMap<Object, Object>();
public Object getAttribute(Object key) {
public Object getAttribute(AttributeKey<Object> key) {
return m_attributes.get(key);
}
public void setAttribute(Object key, Object value) {
public void setAttribute(AttributeKey<Object> key, Object value) {
m_attributes.put(key, value);
}
@@ -106,11 +109,11 @@ public class ProtocolProcessorTest {
AbstractMessage m_receivedMessage;
private Map<Object, Object> m_attributes = new HashMap<Object, Object>();
public Object getAttribute(Object key) {
public Object getAttribute(AttributeKey<Object> key) {
return m_attributes.get(key);
}
public void setAttribute(Object key, Object value) {
public void setAttribute(AttributeKey<Object> 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);