diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 9b93ac8d..abc78ca1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -719,7 +719,8 @@ public class ProtocolProcessor { // ack the client int messageID = msg.variableHeader().messageId(); - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); + MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_MOST_ONCE, + false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.debug("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); diff --git a/tools_scripts/integration/issue410UnsubscribeAckFormat.groovy b/tools_scripts/integration/issue410UnsubscribeAckFormat.groovy new file mode 100644 index 00000000..b728a9be --- /dev/null +++ b/tools_scripts/integration/issue410UnsubscribeAckFormat.groovy @@ -0,0 +1,81 @@ +@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') +@Grab(group='org.eclipse.paho', module='mqtt-client', version='0.4.0') + +import org.eclipse.paho.client.mqttv3.MqttClient +import org.eclipse.paho.client.mqttv3.MqttMessage +import org.eclipse.paho.client.mqttv3.MqttException +import org.eclipse.paho.client.mqttv3.MqttCallback +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence +import org.eclipse.paho.client.mqttv3.MqttConnectOptions +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit + + +class SubscriberCallback implements MqttCallback { + + int m_numReceived = 0 + long m_startTime + boolean firstMessageReceived = false + + private CountDownLatch m_latch = new CountDownLatch(1) + + void waitFinish() { + m_latch.await() + } + + public void connectionLost(Throwable cause) { + m_latch.countDown() + } + + public void messageArrived(String topic, MqttMessage message) throws Exception { + //println "Received message ${new String(message.payload)} on topic [${topic}]" + if (!firstMessageReceived) { + m_startTime = System.currentTimeMillis() + firstMessageReceived = true + } + if (topic == "/exit") { + long stopTime = System.currentTimeMillis() + long spentTime = stopTime - m_startTime + println "/exit received" + println "subscriber disconnected, received ${m_numReceived} messages in ${spentTime} ms" + m_latch.countDown() + } else { + //println "-received ${message} on ${topic} with QoS ${message.qos}" + m_numReceived++ + } + } + + public void deliveryComplete(IMqttDeliveryToken token) { + //To change body of implemented methods use File | Settings | File Templates. + } +} + +if (args.size() < 2) { + println "Usage subscriber " + return +} + +String host = args[0] +int qos = args[1] as int +String tmpDir = System.getProperty("java.io.tmpdir") +MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir) + +int rnd = (Math.random() * 100) as int +MqttClient client = new MqttClient("tcp://${host}:1883", "SubscriberClient${rnd}", dataStore) +def callback = new SubscriberCallback() +client.callback = callback +//Only for Apollo --- +//MqttConnectOptions connOptions = new MqttConnectOptions() +//connOptions.userName ='admin' +//connOptions.password = 'password' as char[] +//client.connect(connOptions) +//Only for Apollo --- +client.connect() +client.subscribe("topic", 0) +println "subscribed to topic" +client.subscribe("/exit", 0) +println "subscribed to /exit" + +callback.waitFinish() +client.diconnect()