From 92a76c9de19c2efb5e8b061d8f78a943b39b601b Mon Sep 17 00:00:00 2001 From: andrea Date: Sat, 20 Jun 2015 12:17:50 +0200 Subject: [PATCH] Fixed issue #38, autoboxing of a nullable Integer given a default to 0 assignment --- .../eclipse/moquette/spi/IMessagesStore.java | 2 +- .../spi/impl/MemoryStorageService.java | 6 ++- .../moquette/spi/impl/ProtocolProcessor.java | 12 ++--- .../spi/impl/events/PublishEvent.java | 4 +- .../spi/impl/storage/StoredPublishEvent.java | 4 +- .../spi/persistence/MapDBPersistentStore.java | 6 ++- .../integration/issue38_pubsubQos2.groovy | 53 +++++++++++++++++++ 7 files changed, 74 insertions(+), 13 deletions(-) create mode 100644 tools_scripts/integration/issue38_pubsubQos2.groovy diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java index 489a77e7..32799d8e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -77,7 +77,7 @@ public interface IMessagesStore { */ List listMessagesInSession(String clientID); - void removeMessageInSession(String clientID, int packetID); + void removeMessageInSession(String clientID, Integer packetID); void dropMessagesInSession(String clientID); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index cdea6f19..400488c1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -102,10 +102,14 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public void removeMessageInSession(String clientID, int messageID) { + public void removeMessageInSession(String clientID, Integer messageID) { List events = m_persistentMessageStore.get(clientID); PublishEvent toRemoveEvt = null; for (PublishEvent evt : events) { + if (evt.getMessageID() == null && messageID == null) { + //was a qos0 message (no ID) + toRemoveEvt = evt; + } if (evt.getMessageID() == messageID) { toRemoveEvt = evt; } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 8f422e4c..6234fdc4 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -208,16 +208,16 @@ class ProtocolProcessor implements EventHandler { if (msg.isPasswordFlag()) { pwd = msg.getPassword(); } else if (!this.allowAnonymous) { - failedCredetials(session); + failedCredentials(session); return; } if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { - failedCredetials(session); + failedCredentials(session); return; } session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername()); } else if (!this.allowAnonymous) { - failedCredetials(session); + failedCredentials(session); return; } @@ -246,7 +246,7 @@ class ProtocolProcessor implements EventHandler { } } - private void failedCredetials(ServerChannel session) { + private void failedCredentials(ServerChannel session) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); session.write(okResp); @@ -311,7 +311,7 @@ class ProtocolProcessor implements EventHandler { if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0 forward2Subscribers(topic, qos, message, retain, messageID); - } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { + } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1 PublishEvent inFlightEvt = new PublishEvent(topic, qos, message, retain, clientID, messageID); //TODO use a message store for TO PUBLISH MESSAGES it has nothing to do with inFlight!! @@ -321,7 +321,7 @@ class ProtocolProcessor implements EventHandler { //NB the PUB_ACK could be sent also after the addInFlight sendPubAck(new PubAckEvent(messageID, clientID)); LOG.debug("replying with PubAck to MSG ID {}", messageID); - } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { + } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 String publishKey = String.format("%s%d", clientID, messageID); //store the message in temp store PublishEvent qos2Persistent = new PublishEvent(topic, qos, message, retain, diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java index 974bc968..1040b82f 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java @@ -31,7 +31,7 @@ public class PublishEvent extends MessagingEvent { boolean m_retain; String m_clientID; //Optional attribute, available only fo QoS 1 and 2 - int m_msgID; + Integer m_msgID; public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, String clientID, Integer msgID) { @@ -65,7 +65,7 @@ public class PublishEvent extends MessagingEvent { return m_clientID; } - public int getMessageID() { + public Integer getMessageID() { return m_msgID; } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java index 483a2bf4..6f0d36b1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java @@ -32,7 +32,7 @@ public class StoredPublishEvent implements Serializable { boolean m_retain; String m_clientID; //Optional attribute, available only fo QoS 1 and 2 - int m_msgID; + Integer m_msgID; public StoredPublishEvent(PublishEvent wrapped) { m_topic = wrapped.getTopic(); @@ -67,7 +67,7 @@ public class StoredPublishEvent implements Serializable { return m_clientID; } - public int getMessageID() { + public Integer getMessageID() { return m_msgID; } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 859ef364..6af9fb9c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -157,13 +157,17 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public void removeMessageInSession(String clientID, int messageID) { + public void removeMessageInSession(String clientID, Integer messageID) { List events = m_persistentMessageStore.get(clientID); if (events == null) { return; } StoredPublishEvent toRemoveEvt = null; for (StoredPublishEvent evt : events) { + if (evt.getMessageID() == null && messageID == null) { + //was a qos0 message (no ID) + toRemoveEvt = evt; + } if (evt.getMessageID() == messageID) { toRemoveEvt = evt; } diff --git a/tools_scripts/integration/issue38_pubsubQos2.groovy b/tools_scripts/integration/issue38_pubsubQos2.groovy new file mode 100644 index 00000000..3a4b3bc0 --- /dev/null +++ b/tools_scripts/integration/issue38_pubsubQos2.groovy @@ -0,0 +1,53 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +String host = args.size() == 1 ? args[0] : "localhost" + +MQTT mqttSub = new MQTT() +mqttSub.setHost(host, 1883) +mqttSub.setClientId("Subscriber(A)") +mqttSub.cleanSession = false +//AT_MOST_ONCE, +//AT_LEAST_ONCE, +//EXACTLY_ONCE +BlockingConnection clientSub = mqttSub.blockingConnection() +clientSub.connect() +println "Sub connected" +Topic[] topics = [new Topic("/sys/abc", QoS.AT_MOST_ONCE)] +byte[] qoses = clientSub.subscribe(topics) +println "Sub subscribed" +clientSub.disconnect() +clientSub.kill() +println "Sub disconnected" + + +MQTT mqttPub = new MQTT() +mqttPub.setHost(host, 1883) +mqttPub.setClientId("Publisher(B)") +mqttPub.cleanSession = false +BlockingConnection clientPub = mqttPub.blockingConnection() +clientPub.connect() +println "Pub connected" +clientPub.publish("/sys/abc", "A message published".bytes, QoS.EXACTLY_ONCE, false) +println "Pub published" + +println "Sub reconnects" +mqttSub = new MQTT() +mqttSub.setHost(host, 1883) +mqttSub.setClientId("Subscriber(A)") +mqttSub.cleanSession = false +clientSub = mqttSub.blockingConnection() +clientSub.connect() +println "Sub connected..and try to receive the message" +Message msg = clientSub.receive() +println "Sub received message \n\t Topic: ${msg.topic}, payload: ${new String(msg.payload)}" +msg.ack() +clientSub.disconnect() //or kill +clientPub.disconnect() + +println "Finished!!\n\n"