Fixed issue #38, autoboxing of a nullable Integer given a default to 0 assignment

broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java

@@ -77,7 +77,7 @@ public interface IMessagesStore {
      */
     List<PublishEvent> listMessagesInSession(String clientID);
     
-    void removeMessageInSession(String clientID, int packetID);
+    void removeMessageInSession(String clientID, Integer packetID);
 
     void dropMessagesInSession(String clientID);
 

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<PublishEvent> 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;
             }

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -208,16 +208,16 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
             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<ValueEvent> {
         }
     }
 
-    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<ValueEvent> {
 
         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<ValueEvent> {
             //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,

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;
     }
 

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;
     }
 }

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<StoredPublishEvent> 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;
             }

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"