Fixed issue #16 publish of will message with qos > 0

ChangeLog.txt

@@ -5,7 +5,8 @@ Version 0.7:
     Fixed bug that didn't update client subscription on persistent store so that on server restart there where ghost subscriptions.
     Fixed Issue #11, multiple republish of persisted QoS1/2 messages on clients reconnections.
     Refactored the publish logic for QoS 1 2 to try to be more readable
-    Added packetIdentifier storarage and generation
+    Added packetIdentifier storage and generation
+    Fixed issue #16, bad packetId for will messages with qos > 0
 
 Version 0.6:
     Fix issue #42, bad password file path composition

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

@@ -47,7 +47,6 @@ import org.eclipse.moquette.proto.messages.SubscribeMessage;
 import org.eclipse.moquette.proto.messages.UnsubAckMessage;
 import org.eclipse.moquette.proto.messages.UnsubscribeMessage;
 import org.eclipse.moquette.server.ConnectionDescriptor;
-import org.eclipse.moquette.server.Constants;
 import org.eclipse.moquette.server.IAuthenticator;
 import org.eclipse.moquette.server.ServerChannel;
 import org.slf4j.Logger;
@@ -130,7 +129,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //init the output ringbuffer
         m_executor = Executors.newFixedThreadPool(1);
 
-        Disruptor<ValueEvent> disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
+        Disruptor<ValueEvent> disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
         disruptor.handleEventsWith(this);
         disruptor.start();
 
@@ -181,7 +180,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, msg.isCleanSession());
         //used to track the client in the subscription and publishing phases.
         session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, msg.getClientID());
-        LOG.debug("Conent create session <{}>", session);
+        LOG.debug("Connect create session <{}>", session);
 
         session.setIdleTime(Math.round(keepAlive * 1.5f));
 
@@ -288,7 +287,6 @@ 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) {
-            String publishKey = String.format("%s%d", clientID, messageID);
             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!!
@@ -323,14 +321,19 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
      * Specialized version to publish will testament message.
      */
     private void forwardPublishWill(WillMessage will, String clientID) {
-        //TODO reimplement it!
-        //TODO it has just to publish the message downstream to the subscribers
+        //it has just to publish the message downstream to the subscribers
         final String topic = will.getTopic();
         final AbstractMessage.QOSType qos = will.getQos();
         final ByteBuffer message = will.getPayload();
         boolean retain = will.isRetained();
         //NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
-        processPublish(clientID, topic, qos, message, retain, 1 /*null*/);
+        if (qos == AbstractMessage.QOSType.MOST_ONE) {
+            forward2Subscribers(topic, qos, message, retain, null);
+        } else {
+            int messageId = m_messagesStore.nextPacketID(clientID);
+            forward2Subscribers(topic, qos, message, retain, messageId);
+        }
+
     }
     
     /**

tools_scripts/integration/issue16_will_publish.groovy

@@ -0,0 +1,51 @@
+@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/')
+//@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.1', ext='jar')
+@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 mqttWill = new MQTT()
+mqttWill.setHost(host, 1883)
+mqttWill.setClientId("WillClient")
+mqttWill.setWillTopic("/willtopic")
+mqttWill.setWillMessage("Will message testment")
+//AT_MOST_ONCE,
+//AT_LEAST_ONCE,
+//EXACTLY_ONCE
+mqttWill.willQos = QoS.AT_LEAST_ONCE
+BlockingConnection willClient = mqttWill.blockingConnection()
+willClient.connect()
+println "Will Client connected"
+
+
+println "Connect a subscriber to the topic"
+MQTT mqtt = new MQTT()
+//mqtt.setHost("test.mosquitto.org", 1883);
+mqtt.setHost(host, 1883)
+mqtt.setCleanSession(true)
+mqtt.setHost(host, 1883)
+
+mqtt.setClientId("SubscriberClient")
+BlockingConnection connection = mqtt.blockingConnection()
+connection.connect()
+println "Connected to ${host}"
+
+Topic[] topics = [new Topic("/willtopic", QoS.AT_MOST_ONCE)]
+byte[] qoses = connection.subscribe(topics)
+println "subscribed to /willtopic qos: ${qoses[0]}"
+
+//simulate a disconnection of the willclient
+willClient.kill()
+
+Message message = connection.receive()
+assert message.topic == "/willtopic"
+message.ack()
+
+connection.disconnect()
+