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

Removed some unusefull PublishEvent constructros

Browse Source
This commit is contained in:
selva.andre
2014-08-18 23:27:08 +02:00
parent aee85a7cb0
commit 5c5e640b04
2 changed files with 32 additions and 75 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
+28 -57
View File
@@ -280,34 +280,41 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
final AbstractMessage.QOSType qos = msg.getQos(); final AbstractMessage.QOSType qos = msg.getQos();
final ByteBuffer message = msg.getPayload(); final ByteBuffer message = msg.getPayload();
boolean retain = msg.isRetainFlag(); boolean retain = msg.isRetainFlag();
processPublish(clientID, topic, qos, message, retain, msg.getMessageID());
}
private void processPublish(String clientID, String topic, QOSType qos, ByteBuffer message, boolean retain, Integer messageID) {
LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}",
clientID, topic, qos); clientID, topic, qos);
String publishKey = null; String publishKey = null;
if (qos == AbstractMessage.QOSType.LEAST_ONE) { if (qos == AbstractMessage.QOSType.LEAST_ONE) {
//store the temporary message //store the temporary message
publishKey = String.format("%s%d", clientID, msg.getMessageID()); publishKey = String.format("%s%d", clientID, messageID);
m_messagesStore.addInFlight(new PublishEvent(msg, clientID), publishKey); PublishEvent inFlightEvt = new PublishEvent(topic, qos, message, retain,
clientID, messageID);
m_messagesStore.addInFlight(inFlightEvt, publishKey);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
publishKey = String.format("%s%d", clientID, msg.getMessageID()); publishKey = String.format("%s%d", clientID, messageID);
//store the message in temp store //store the message in temp store
m_messagesStore.persistQoS2Message(publishKey, new PublishEvent(msg, clientID)); PublishEvent qos2Persistent = new PublishEvent(topic, qos, message, retain,
sendPubRec(clientID, msg.getMessageID()); clientID, messageID);
m_messagesStore.persistQoS2Message(publishKey, qos2Persistent);
sendPubRec(clientID, messageID);
} }
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) { if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) {
publish2Subscribers(topic, qos, message, retain, msg.getMessageID()); publish2Subscribers(topic, qos, message, retain, messageID);
} }
if (qos == AbstractMessage.QOSType.LEAST_ONE) { if (qos == AbstractMessage.QOSType.LEAST_ONE) {
if (publishKey == null) { if (publishKey == null) {
throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + msg); throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + messageID);
} }
m_messagesStore.cleanInFlight(publishKey); m_messagesStore.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(msg.getMessageID(), clientID)); sendPubAck(new PubAckEvent(messageID, clientID));
LOG.debug("replying with PubAck to MSG ID {}", msg.getMessageID()); LOG.debug("replying with PubAck to MSG ID {}", messageID);
} }
if (retain) { if (retain) {
@@ -320,51 +327,15 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
} }
} }
@Deprecated /**
protected void processPublish(PublishEvent evt) { * Specialized version to publish will testament message.
LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", evt); */
final String topic = evt.getTopic(); private void processPublish(WillMessage will, String clientID) {
final AbstractMessage.QOSType qos = evt.getQos(); final String topic = will.getTopic();
final ByteBuffer message = evt.getMessage(); final AbstractMessage.QOSType qos = will.getQos();
boolean retain = evt.isRetain(); final ByteBuffer message = will.getPayload();
boolean retain = will.isRetained();
LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", processPublish(clientID, topic, qos, message, retain, null);
evt.getClientID(), evt.getTopic(), evt.getQos());
String publishKey = null;
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
//store the temporary message
publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID());
m_messagesStore.addInFlight(evt, publishKey);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID());
//store the message in temp store
m_messagesStore.persistQoS2Message(publishKey, evt);
sendPubRec(evt.getClientID(), evt.getMessageID());
}
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) {
publish2Subscribers(topic, qos, message, retain, evt.getMessageID());
}
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
if (publishKey == null) {
throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + evt);
}
m_messagesStore.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
LOG.debug("replying with PubAck to MSG ID {}", evt.getMessageID());
}
if (retain) {
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, message, qos);
}
}
} }
/** /**
@@ -568,9 +539,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//TODO publish the Will message (if any) for the clientID //TODO publish the Will message (if any) for the clientID
if (m_willStore.containsKey(clientID)) { if (m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID); WillMessage will = m_willStore.get(clientID);
PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(), // PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(),
will.getPayload(), will.isRetained(), clientID); // will.getPayload(), will.isRetained(), clientID);
processPublish(pubEvt); processPublish(will, clientID);
m_willStore.remove(clientID); m_willStore.remove(clientID);
} }
} }
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+4 -18
View File
@@ -34,30 +34,16 @@ public class PublishEvent extends MessagingEvent {
//Optional attribute, available only fo QoS 1 and 2 //Optional attribute, available only fo QoS 1 and 2
int m_msgID; int m_msgID;
public PublishEvent(PublishMessage pubMsg, String clientID) {
m_topic = pubMsg.getTopicName();
m_qos = pubMsg.getQos();
m_message = pubMsg.getPayload();
m_retain = pubMsg.isRetainFlag();
m_clientID = clientID;
if (pubMsg.getQos() != QOSType.MOST_ONE) {
m_msgID = pubMsg.getMessageID();
}
}
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID) { String clientID, Integer msgID) {
m_topic = topic; m_topic = topic;
m_qos = qos; m_qos = qos;
m_message = message; m_message = message;
m_retain = retain; m_retain = retain;
m_clientID = clientID; m_clientID = clientID;
} if (qos != QOSType.MOST_ONE) {
m_msgID = msgID;
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, }
String clientID, int msgID) {
this(topic, qos, message, retain, clientID);
m_msgID = msgID;
} }
public String getTopic() { public String getTopic() {