Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

This closes #3537

This commit is contained in:
Justin Bertram 2021-05-25 11:24:54 -05:00
parent a3fb3ffdce 6ec7a69d45
commit 67949b9484
2 changed files with 51 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSubscriptionManager.java
View File

@ -196,6 +196,7 @@ public class MQTTSubscriptionManager {
SimpleString internalQueueName = getQueueNameForTopic(internalAddress); SimpleString internalQueueName = getQueueNameForTopic(internalAddress);
session.getSessionState().removeSubscription(address); session.getSessionState().removeSubscription(address);
Queue queue = session.getServer().locateQueue(internalQueueName);
SimpleString sAddress = SimpleString.toSimpleString(internalAddress); SimpleString sAddress = SimpleString.toSimpleString(internalAddress);
AddressInfo addressInfo = session.getServerSession().getAddress(sAddress); AddressInfo addressInfo = session.getServerSession().getAddress(sAddress);
if (addressInfo != null && addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST)) { if (addressInfo != null && addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST)) {
@ -207,7 +208,6 @@ public class MQTTSubscriptionManager {
} }
} else { } else {
consumers.remove(address); consumers.remove(address);
Queue queue = session.getServer().locateQueue(internalQueueName);
Set<Consumer> queueConsumers; Set<Consumer> queueConsumers;
if (queue != null && (queueConsumers = (Set<Consumer>) queue.getConsumers()) != null) { if (queue != null && (queueConsumers = (Set<Consumer>) queue.getConsumers()) != null) {
for (Consumer consumer : queueConsumers) { for (Consumer consumer : queueConsumers) {
@ -217,10 +217,16 @@ public class MQTTSubscriptionManager {
} }
} }
if (session.getServerSession().executeQueueQuery(internalQueueName).isExists()) { if (queue != null) {
assert session.getServerSession().executeQueueQuery(internalQueueName).isExists();
if (queue.isConfigurationManaged()) {
queue.deleteAllReferences();
} else {
session.getServerSession().deleteQueue(internalQueueName); session.getServerSession().deleteQueue(internalQueueName);
} }
} }
}
private SimpleString getQueueNameForTopic(String topic) { private SimpleString getQueueNameForTopic(String topic) {
return new SimpleString(session.getSessionState().getClientId() + "." + topic); return new SimpleString(session.getSessionState().getClientId() + "." + topic);

42
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTQueueCleanTest.java
View File

@ -17,7 +17,10 @@
package org.apache.activemq.artemis.tests.integration.mqtt.imported; package org.apache.activemq.artemis.tests.integration.mqtt.imported;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.tests.util.Wait; import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.Test; import org.junit.Test;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -31,6 +34,45 @@ public class MQTTQueueCleanTest extends MQTTTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(MQTTQueueCleanTest.class); private static final Logger LOG = LoggerFactory.getLogger(MQTTQueueCleanTest.class);
@Test
public void testQueueClean() throws Exception {
testQueueClean(false);
}
@Test
public void testManagedQueueClean() throws Exception {
testQueueClean(true);
}
private void testQueueClean(boolean managed) throws Exception {
String address = "clean/test";
String clientId = "mqtt-client";
String queueName = "::mqtt-client.clean.test";
if (managed) {
server.addAddressInfo(new AddressInfo(address)
.addRoutingType(RoutingType.MULTICAST));
server.createQueue(new QueueConfiguration(queueName)
.setAddress(address)
.setRoutingType(RoutingType.MULTICAST)
.setConfigurationManaged(true));
}
MQTTClientProvider clientProvider = getMQTTClientProvider();
clientProvider.setClientId(clientId);
initializeConnection(clientProvider);
clientProvider.subscribe(address, AT_LEAST_ONCE);
clientProvider.disconnect();
if (managed) {
assertTrue(Wait.waitFor(() -> server.locateQueue(SimpleString.toSimpleString(queueName)) != null &&
server.locateQueue(SimpleString.toSimpleString(queueName)).getConsumerCount() == 0, 5000, 10));
} else {
assertTrue(Wait.waitFor(() -> server.locateQueue(SimpleString.toSimpleString(queueName)) == null, 5000, 10));
}
}
@Test @Test
public void testQueueCleanWhenConnectionSynExeConnectAndDisconnect() throws Exception { public void testQueueCleanWhenConnectionSynExeConnectAndDisconnect() throws Exception {
Random random = new Random(); Random random = new Random();