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

Extracted MQTT session constants into separate class

Browse Source
This commit is contained in:
selva.andre
2012-09-01 10:10:48 +02:00
parent 65e34ddc40
commit 04872e07f7
2 changed files with 21 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java
+9
View File
@@ -0,0 +1,9 @@
package org.dna.mqtt.moquette.server;
/**
* Server constants keeper
*/
public class Constants {
protected static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession";
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+12 -13
View File
@@ -24,7 +24,6 @@ import org.slf4j.LoggerFactory;
*/
public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected static final String ATTR_CLIENTID = "ClientID";
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
/**
* Maps CLIENT_ID to the IoSession that represents the connection
@@ -87,7 +86,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
if (m_clientIDs.containsKey(msg.getClientID())) {
//clean the subscriptions if the old used a cleanSession = true
IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute("cleanSession");
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_messaging.removeSubscriptions(msg.getClientID());
@@ -104,9 +103,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
int keepAlive = msg.getKeepAlive();
session.setAttribute("keepAlive", keepAlive);
session.setAttribute("cleanSession", msg.isCleanSession());
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(ATTR_CLIENTID, msg.getClientID());
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
@@ -114,7 +113,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
if (msg.isWillFlag()) {
QOSType willQos = QOSType.values()[msg.getWillQos()];
m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(),
willQos, msg.isWillRetain(), msg.getClientID());
willQos, msg.isWillRetain(), msg.getClientID(), session);
}
//handle user authentication
@@ -148,7 +147,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected void handleSubscribe(IoSession session, SubscribeMessage msg) {
LOG.debug("handleSubscribe, registering the subscriptions");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
m_messaging.subscribe((String) session.getAttribute(ATTR_CLIENTID),
m_messaging.subscribe((String) session.getAttribute(Constants.ATTR_CLIENTID),
req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]);
}
@@ -167,7 +166,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) {
LOG.info("unregistering the subscriptions");
for (String topic : msg.topics()) {
m_messaging.unsubscribe(topic, (String) session.getAttribute(ATTR_CLIENTID));
m_messaging.unsubscribe(topic, (String) session.getAttribute(Constants.ATTR_CLIENTID));
}
//ack the client
UnsubAckMessage ackMessage = new UnsubAckMessage();
@@ -178,19 +177,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
protected void handlePublish(IoSession session, PublishMessage message) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
if (message.getQos() == QOSType.MOST_ONE) {
m_messaging.publish(message.getTopicName(), message.getPayload(),
message.getQos(), message.isRetainFlag(), clientID);
message.getQos(), message.isRetainFlag(), clientID, session);
} else {
m_messaging.publish(message.getTopicName(), message.getPayload(),
message.getQos(), message.isRetainFlag(), clientID, message.getMessageID());
message.getQos(), message.isRetainFlag(), clientID, message.getMessageID(), session);
}
}
protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
//remove from clientIDs
// m_clientIDsLock.lock();
// try {
@@ -198,7 +197,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
// } finally {
// m_clientIDsLock.unlock();
// }
boolean cleanSession = (Boolean) session.getAttribute("cleanSession");
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_messaging.removeSubscriptions(clientID);
@@ -252,7 +251,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
public void disconnect(IoSession session) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_clientIDsLock.lock();
try {
m_clientIDs.remove(clientID);