diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java index 8f4aab8c..bacd098d 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java @@ -321,7 +321,7 @@ final class MQTTConnection { } switch (qos) { case AT_MOST_ONCE: - postOffice.receivedPublishQos0(topic, username, clientId, payload, retain); + postOffice.receivedPublishQos0(topic, username, clientId, payload, retain, msg); break; case AT_LEAST_ONCE: { final int messageID = msg.variableHeader().packetId(); diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index beee1cbf..60d4fdea 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/broker/src/main/java/io/moquette/broker/PostOffice.java @@ -143,7 +143,7 @@ class PostOffice { mqttConnection.sendUnsubAckMessage(topics, clientID, messageId); } - void receivedPublishQos0(Topic topic, String username, String clientID, ByteBuf payload, boolean retain) { + void receivedPublishQos0(Topic topic, String username, String clientID, ByteBuf payload, boolean retain, MqttPublishMessage msg) { if (!authorizator.canWrite(topic, username, clientID)) { LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic: {}", clientID, topic); return; @@ -155,9 +155,6 @@ class PostOffice { retainedRepository.cleanRetained(topic); } - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, AT_MOST_ONCE, false, 0); - MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic.toString(), 0); - MqttPublishMessage msg = new MqttPublishMessage(fixedHeader, varHeader, payload.retainedDuplicate()); interceptor.notifyTopicPublished(msg, clientID, username); } diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java index 5a13ff8e..ec4e8f5a 100644 --- a/broker/src/main/java/io/moquette/broker/Session.java +++ b/broker/src/main/java/io/moquette/broker/Session.java @@ -4,6 +4,7 @@ import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.*; +import io.netty.util.ReferenceCountUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -335,12 +336,13 @@ class Session { public void receivedPublishQos2(int messageID, MqttPublishMessage msg) { qos2Receiving.put(messageID, msg); - msg.retain(); + msg.retain(); // retain to put in the inflight map mqttConnection.sendPublishReceived(messageID); } public void receivedPubRelQos2(int messageID) { - qos2Receiving.remove(messageID); + final MqttPublishMessage removedMsg = qos2Receiving.remove(messageID); + ReferenceCountUtil.release(removedMsg); } @Override diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java index f5777dcc..e181ef59 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java @@ -42,10 +42,18 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { @Override public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception { - logMQTTMessage(ctx, message, "C->B"); + logMQTTMessageRead(ctx, message); ctx.fireChannelRead(message); } + private void logMQTTMessageRead(ChannelHandlerContext ctx, Object message) throws Exception { + logMQTTMessage(ctx, message, "C->B"); + } + + private void logMQTTMessageWrite(ChannelHandlerContext ctx, Object message) throws Exception { + logMQTTMessage(ctx, message, "C<-B"); + } + private void logMQTTMessage(ChannelHandlerContext ctx, Object message, String direction) throws Exception { if (!(message instanceof MqttMessage)) { return; @@ -106,7 +114,7 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { - logMQTTMessage(ctx, msg, "C<-B"); + logMQTTMessageWrite(ctx, msg); ctx.write(msg, promise).addListener(CLOSE_ON_FAILURE); } } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index 4b7cbd15..80a9948b 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java @@ -91,7 +91,12 @@ public class PostOfficePublishTest { // Exercise final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, FAKE_CLIENT_ID, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, FAKE_CLIENT_ID, payload, false, + MqttMessageBuilders.publish() + .payload(payload.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); // Verify ConnectionTestUtils.verifyReceivePublish(channel, NEWS_TOPIC, "Hello world!"); @@ -187,7 +192,12 @@ public class PostOfficePublishTest { // Exercise final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, FAKE_CLIENT_ID, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, FAKE_CLIENT_ID, payload, false, + MqttMessageBuilders.publish() + .payload(payload.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); // Verify ConnectionTestUtils.verifyReceivePublish(channel1, NEWS_TOPIC, "Hello world!"); @@ -206,7 +216,12 @@ public class PostOfficePublishTest { // Exercise final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, FAKE_CLIENT_ID, anyPayload, true); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, FAKE_CLIENT_ID, anyPayload, true, + MqttMessageBuilders.publish() + .payload(anyPayload) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); // Verify assertTrue("QoS0 MUST clean retained message for topic", retainedRepository.isEmtpy()); @@ -363,7 +378,12 @@ public class PostOfficePublishTest { // publish a QoS0 retained message // Exercise final ByteBuf qos0Payload = Unpooled.copiedBuffer("QoS0 payload", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, connection.getClientId(), qos0Payload, true); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, connection.getClientId(), qos0Payload, true, + MqttMessageBuilders.publish() + .payload(qos0Payload) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); // Verify assertTrue("Retained message for topic /news must be cleared", retainedRepository.isEmtpy()); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java index 0041403e..739d6c92 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java @@ -1,8 +1,6 @@ package io.moquette.broker; -import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.MockAuthenticator; import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy; import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory; @@ -225,7 +223,12 @@ public class PostOfficeSubscribeTest { assertEquals("After a reconnect, subscription MUST be still present", 1, subscriptions.size()); final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false, + MqttMessageBuilders.publish() + .payload(payload.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyPublishIsReceived(anotherChannel, AT_MOST_ONCE, "Hello world!"); } @@ -265,7 +268,12 @@ public class PostOfficeSubscribeTest { // publish on /news final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false, + MqttMessageBuilders.publish() + .payload(payload) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); // verify no publish is fired ConnectionTestUtils.verifyNoPublishIsReceived(channel); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java index ac01e840..b46005dd 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java @@ -124,7 +124,12 @@ public class PostOfficeUnsubscribeTest { // publish on /news final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false, + MqttMessageBuilders.publish() + .payload(payload.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyPublishIsReceived(channel, AT_MOST_ONCE, "Hello world!"); @@ -132,7 +137,12 @@ public class PostOfficeUnsubscribeTest { // publish on /news final ByteBuf payload2 = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload2, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload2, false, + MqttMessageBuilders.publish() + .payload(payload) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyNoPublishIsReceived(channel); } @@ -152,7 +162,12 @@ public class PostOfficeUnsubscribeTest { subscribe(connection, NEWS_TOPIC, AT_MOST_ONCE); // publish on /news final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false, + MqttMessageBuilders.publish() + .payload(payload.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyPublishIsReceived(channel, AT_MOST_ONCE, "Hello world!"); @@ -167,7 +182,12 @@ public class PostOfficeUnsubscribeTest { // publish on /news final ByteBuf payload2 = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload2, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload2, false, + MqttMessageBuilders.publish() + .payload(payload2) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyNoPublishIsReceived(anotherChannel); } @@ -242,7 +262,12 @@ public class PostOfficeUnsubscribeTest { subscribe(connection, NEWS_TOPIC, AT_MOST_ONCE); // publish on /news final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload, false, + MqttMessageBuilders.publish() + .payload(payload.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyPublishIsReceived((EmbeddedChannel) connection.channel, AT_MOST_ONCE, "Hello world!"); @@ -257,7 +282,12 @@ public class PostOfficeUnsubscribeTest { subscribe(subscriberConnection, NEWS_TOPIC, AT_MOST_ONCE); // publish on /news final ByteBuf payload2 = Unpooled.copiedBuffer("Hello world2!", Charset.defaultCharset()); - sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload2, false); + sut.receivedPublishQos0(new Topic(NEWS_TOPIC), TEST_USER, TEST_PWD, payload2, false, + MqttMessageBuilders.publish() + .payload(payload2.retainedDuplicate()) + .qos(MqttQoS.AT_MOST_ONCE) + .retained(false) + .topicName(NEWS_TOPIC).build()); ConnectionTestUtils.verifyPublishIsReceived(subscriberChannel, AT_MOST_ONCE, "Hello world2!"); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index d966a267..13c8f69b 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -192,187 +192,4 @@ public class ServerIntegrationPahoTest { MqttMessage messageOnB = cbSubscriber2.waitMessage(1); assertEquals("Hello", new String(messageOnB.getPayload(), UTF_8)); } - -// @Test -// public void testForceClientDisconnection_issue116() throws Exception { -// LOG.info("*** testForceClientDisconnection_issue118 ***"); -// MessageCollector cbSubscriber1 = new MessageCollector(); -// MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1); -// LOG.info("Connected 'subscriber' first time"); -// clientXA.subscribe("topic", 0); -// LOG.info("Subscribed 'topic' from 'subscriber' first time"); -// -// MqttClient clientXB = createClient("publisher", "X"); -// LOG.info("Connected 'publisher' first time"); -// clientXB.publish("topic", "Hello".getBytes(), 2, false); -// LOG.info("Published on 'topic' from 'publisher' first time"); -// -// LOG.info("Creating second new 'subscriber'"); -// MessageCollector cbSubscriber2 = new MessageCollector(); -// MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2); -// LOG.info("Connected 'subscriber' second time"); -// clientYA.subscribe("topic", 0); -// -// MqttClient clientYB = createClient("publisher", "Y"); -// clientYB.publish("topic", "Hello 2".getBytes(), 2, true); -// -// // Verify that the second subscriber client get notified and not the first. -// assertTrue(cbSubscriber1.connectionLost()); -// assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload())); -// } - -// @Test -// public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws Exception { -// LOG.info("*** testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142 ***"); -// MqttConnectOptions options = new MqttConnectOptions(); -// options.setCleanSession(false); -// m_client.connect(options); -// m_client.subscribe("/topic", 0); -// m_client.publish("/topic", "Hello".getBytes(), 0, true); -// m_client.disconnect(); -// -// // second loop -// m_client.connect(options); -// m_client.subscribe("/topic", 0); -// m_client.publish("/topic", "Hello".getBytes(), 0, true); -// m_client.disconnect(); // this should give timeout -// -// assertFalse("after a disconnect the client should be disconnected", m_client.isConnected()); -// } - -// @Test -// public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception { -// LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***"); -// // MessageCollector m_messagesCollector = new MessageCollector(); -// // m_client.setCallback(m_messagesCollector); -// m_publisher.connect(); -// -// MqttConnectOptions options = new MqttConnectOptions(); -// options.setCleanSession(false); -// m_client.connect(options); -// m_client.subscribe("/topic", 1); -// -// // force the publisher to send -// m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false); -// -// // read the first message and drop the connection -// MqttMessage msg = m_messagesCollector.waitMessage(1); -// assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); -// m_client.disconnect(); -// -// m_publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), 1, false); -// m_publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), 1, false); -// -// // reconnect and expect to receive the hello 2 message -// m_client.connect(options); -// msg = m_messagesCollector.waitMessage(1); -// assertEquals("Hello world MQTT!!-2", new String(msg.getPayload())); -// -// msg = m_messagesCollector.waitMessage(1); -// assertEquals("Hello world MQTT!!-3", new String(msg.getPayload())); -// m_client.disconnect(); -// } - -// @Test -// public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception { -// LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***"); -// MessageCollector messagesCollector = new MessageCollector(); -// m_client.setCallback(messagesCollector); -// m_publisher.connect(); -// -// MqttConnectOptions options = new MqttConnectOptions(); -// options.setCleanSession(false); -// m_client.connect(options); -// m_client.subscribe("/topic", 1); -// m_client.disconnect(); -// -// // force the publisher to send -// m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false); -// -// // reconnect and expect to receive the hello 2 message -// m_client.connect(options); -// m_client.subscribe("/topic", 1); -// MqttMessage msg = messagesCollector.waitMessage(1); -// assertNotNull(msg); -// assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); -// } - -// /** -// * subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and -// * send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1" -// * "hello2" -// */ -// @Test -// public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws Exception { -// LOG.info("*** checkQoS2SubscriberDisconnectReceivePersistedPublishes ***"); -// MessageCollector messagesCollector = new MessageCollector(); -// m_client.setCallback(messagesCollector); -// MqttConnectOptions options = new MqttConnectOptions(); -// options.setCleanSession(false); -// m_client.connect(options); -// m_client.subscribe("topic", 2); -// m_client.disconnect(); -// -// m_publisher.connect(); -// m_publisher.publish("topic", "Hello1".getBytes(), 2, true); -// m_publisher.publish("topic", "Hello2".getBytes(), 2, true); -// m_publisher.disconnect(); -// -// // subscriber reconnects -// m_client.connect(options); -// // subscriber should receive the 2 messages missed -// MqttMessage msg = messagesCollector.waitMessage(1); -// assertNotNull(msg); -// assertEquals("Hello1", new String(msg.getPayload())); -// msg = messagesCollector.waitMessage(1); -// assertNotNull(msg); -// assertEquals("Hello2", new String(msg.getPayload())); -// } -// -// /** -// * subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a -// * message "hello" on "a/b" subscriber must receive only a single message not twice -// */ -// @Test -// public void checkSinglePublishOnOverlappingSubscriptions() throws Exception { -// LOG.info("*** checkSinglePublishOnOverlappingSubscriptions ***"); -// MessageCollector messagesCollector = new MessageCollector(); -// m_client.setCallback(messagesCollector); -// m_publisher.connect(); -// -// MqttConnectOptions options = new MqttConnectOptions(); -// options.setCleanSession(false); -// m_client.connect(options); -// m_client.subscribe("a/+", 2); -// m_client.subscribe("a/b", 1); -// -// // force the publisher to send -// m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), 1, false); -// -// // reconnect and expect to receive the hello 2 message -// MqttMessage msg = messagesCollector.waitMessage(1); -// assertEquals("Hello world MQTT!!", new String(msg.getPayload())); -// -// // try to listen a second publish -// msg = messagesCollector.waitMessage(1); -// assertNull(msg); -// } - - protected MqttClient createClient(String clientName, String storeSuffix) throws MqttException { - return createClient(clientName, storeSuffix, null); - } - - protected MqttClient createClient(String clientName, String storeSuffix, MessageCollector cb) throws MqttException { - String tmpDir = System.getProperty("java.io.tmpdir"); - // clientX connect and subscribe to /topic QoS2 - MqttClientPersistence dsClient = new MqttDefaultFilePersistence( - tmpDir + File.separator + storeSuffix + clientName); - MqttClient client = new MqttClient("tcp://localhost:1883", clientName, dsClient); - if (cb != null) { - client.setCallback(cb); - } - client.connect(); - return client; - } - }