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

Used annotation dispatch for some protocol messages

Browse Source
This commit is contained in:
selva.andre
2014-08-17 16:47:14 +02:00
parent 43677cca89
commit 5446d25528
3 changed files with 44 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+26 -7
View File
@@ -49,6 +49,7 @@ import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.IAuthenticator;
@@ -252,7 +253,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
}
void processPubAck(String clientID, int messageID) {
@MQTTMessage(message = PubAckMessage.class)
void processPubAck(ServerChannel session, PubAckMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = msg.getMessageID();
//Remove the message from message store
m_messagesStore.cleanPersistedPublishMessage(clientID, messageID);
}
@@ -428,7 +432,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
* to all interested subscribers.
* */
void processPubRel(String clientID, int messageID) {
@MQTTMessage(message = PubRelMessage.class)
void processPubRel(ServerChannel session, PubRelMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_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);
PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey);
@@ -456,7 +463,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubCompMessage));
}
void processPubRec(String clientID, int messageID) {
@MQTTMessage(message = PubRecMessage.class)
void processPubRec(ServerChannel session, PubRecMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_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);
PubRelMessage pubRelMessage = new PubRelMessage();
@@ -467,7 +477,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage));
}
void processPubComp(String clientID, int messageID) {
@MQTTMessage(message = PubCompMessage.class)
void processPubComp(ServerChannel session, PubCompMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_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
String publishKey = String.format("%s%d", clientID, messageID);
@@ -513,7 +526,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
void processUnsubscribe(ServerChannel session, String clientID, List<String> topics, int messageID) {
@MQTTMessage(message = UnsubscribeMessage.class)
void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
List<String> topics = msg.topicFilters();
int messageID = msg.getMessageID();
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
LOG.debug("processUnsubscribe invoked, removing subscription on topics {}, for clientID <{}>", topics, clientID);
for (String topic : topics) {
@@ -527,8 +544,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
session.write(ackMessage);
}
void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) {
@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);
LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID());
for (SubscribeMessage.Couple req : msg.subscriptions()) {
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+7 -32
View File
@@ -150,19 +150,18 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (evt instanceof ProtocolEvent) {
ServerChannel session = ((ProtocolEvent) evt).getSession();
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
if (message instanceof ConnectMessage) {
// m_processor.processConnect(session, (ConnectMessage) message);
if (message instanceof ConnectMessage ||
message instanceof UnsubscribeMessage ||
message instanceof SubscribeMessage ||
message instanceof PubRelMessage ||
message instanceof PubRecMessage ||
message instanceof PubCompMessage ||
message instanceof PubAckMessage) {
annotationHelper.dispatch(session, message);
} else if (message instanceof PublishMessage) {
PublishEvent pubEvt;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
pubEvt = new PublishEvent((PublishMessage) message, clientID);
// if (message.getQos() == QOSType.MOST_ONE) {
// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
//
// } else {
// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
// }
m_processor.processPublish(pubEvt);
} else if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
@@ -171,30 +170,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
//close the TCP connection
//session.close(true);
m_processor.processDisconnect(session, clientID, cleanSession);
} else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_processor.processUnsubscribe(session, clientID, unsubMsg.topicFilters(), unsubMsg.getMessageID());
} else if (message instanceof SubscribeMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_processor.processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession);
} else if (message instanceof PubRelMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRelMessage) message).getMessageID();
m_processor.processPubRel(clientID, messageID);
} else if (message instanceof PubRecMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRecMessage) message).getMessageID();
m_processor.processPubRec(clientID, messageID);
} else if (message instanceof PubCompMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubCompMessage) message).getMessageID();
m_processor.processPubComp(clientID, messageID);
} else if (message instanceof PubAckMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubAckMessage) message).getMessageID();
m_processor.processPubAck(clientID, messageID);
} else {
throw new RuntimeException("Illegal message received " + message);
}