From 5c5e640b044a75f71bc0093fdafa732da4b7c26e Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Mon, 18 Aug 2014 23:27:08 +0200 Subject: [PATCH] Removed some unusefull PublishEvent constructros --- .../messaging/spi/impl/ProtocolProcessor.java | 85 ++++++------------- .../spi/impl/events/PublishEvent.java | 22 +---- 2 files changed, 32 insertions(+), 75 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 722fa657..452ab740 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -280,34 +280,41 @@ class ProtocolProcessor implements EventHandler { final AbstractMessage.QOSType qos = msg.getQos(); final ByteBuffer message = msg.getPayload(); 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 {}", clientID, topic, qos); String publishKey = null; if (qos == AbstractMessage.QOSType.LEAST_ONE) { //store the temporary message - publishKey = String.format("%s%d", clientID, msg.getMessageID()); - m_messagesStore.addInFlight(new PublishEvent(msg, clientID), publishKey); + publishKey = String.format("%s%d", clientID, messageID); + PublishEvent inFlightEvt = new PublishEvent(topic, qos, message, retain, + clientID, messageID); + m_messagesStore.addInFlight(inFlightEvt, publishKey); } 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 - m_messagesStore.persistQoS2Message(publishKey, new PublishEvent(msg, clientID)); - sendPubRec(clientID, msg.getMessageID()); + PublishEvent qos2Persistent = new PublishEvent(topic, qos, message, retain, + clientID, messageID); + m_messagesStore.persistQoS2Message(publishKey, qos2Persistent); + sendPubRec(clientID, messageID); } //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher 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 (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); - sendPubAck(new PubAckEvent(msg.getMessageID(), clientID)); - LOG.debug("replying with PubAck to MSG ID {}", msg.getMessageID()); + sendPubAck(new PubAckEvent(messageID, clientID)); + LOG.debug("replying with PubAck to MSG ID {}", messageID); } if (retain) { @@ -320,51 +327,15 @@ class ProtocolProcessor implements EventHandler { } } - @Deprecated - protected void processPublish(PublishEvent evt) { - LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", evt); - final String topic = evt.getTopic(); - final AbstractMessage.QOSType qos = evt.getQos(); - final ByteBuffer message = evt.getMessage(); - boolean retain = evt.isRetain(); - - LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", - 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); - } - } + /** + * Specialized version to publish will testament message. + */ + private void processPublish(WillMessage will, String clientID) { + final String topic = will.getTopic(); + final AbstractMessage.QOSType qos = will.getQos(); + final ByteBuffer message = will.getPayload(); + boolean retain = will.isRetained(); + processPublish(clientID, topic, qos, message, retain, null); } /** @@ -568,9 +539,9 @@ class ProtocolProcessor implements EventHandler { //TODO publish the Will message (if any) for the clientID if (m_willStore.containsKey(clientID)) { WillMessage will = m_willStore.get(clientID); - PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(), - will.getPayload(), will.isRetained(), clientID); - processPublish(pubEvt); +// PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(), +// will.getPayload(), will.isRetained(), clientID); + processPublish(will, clientID); m_willStore.remove(clientID); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 1cd4df4a..6cee0313 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -34,30 +34,16 @@ public class PublishEvent extends MessagingEvent { //Optional attribute, available only fo QoS 1 and 2 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, - String clientID) { + String clientID, Integer msgID) { m_topic = topic; m_qos = qos; m_message = message; m_retain = retain; m_clientID = clientID; - } - - public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, - String clientID, int msgID) { - this(topic, qos, message, retain, clientID); - m_msgID = msgID; + if (qos != QOSType.MOST_ONE) { + m_msgID = msgID; + } } public String getTopic() {