diff --git a/ChangeLog.txt b/ChangeLog.txt index 6a14618d..63e92647 100644 --- a/ChangeLog.txt +++ b/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 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 ce01f816..be4cc2de 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 @@ -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 { //init the output ringbuffer m_executor = Executors.newFixedThreadPool(1); - Disruptor disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); + Disruptor disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); disruptor.handleEventsWith(this); disruptor.start(); @@ -181,7 +180,7 @@ class ProtocolProcessor implements EventHandler { 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 { 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 { * 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); + } + } /** diff --git a/tools_scripts/integration/issue16_will_publish.groovy b/tools_scripts/integration/issue16_will_publish.groovy new file mode 100644 index 00000000..dd1fbee9 --- /dev/null +++ b/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() +