From 2f2519409f0bc9576f1947f88ecb610bcead1603 Mon Sep 17 00:00:00 2001 From: Anmol Saxena Date: Wed, 22 Apr 2026 18:26:12 +0530 Subject: [PATCH 1/3] ARTEMIS-3164:Support prefix to create temporary resources. --- .../amqp/broker/ProtonProtocolManager.java | 23 ++ .../openwire/OpenWireProtocolManager.java | 23 ++ .../core/protocol/stomp/StompConnection.java | 13 +- .../protocol/stomp/StompProtocolManager.java | 6 +- .../core/protocol/stomp/StompSession.java | 53 ++- .../stomp/VersionedStompFrameHandler.java | 17 +- .../core/impl/CoreProtocolManager.java | 23 ++ .../protocol/AbstractProtocolManager.java | 23 ++ .../spi/core/protocol/ProtocolManager.java | 6 + .../stomp/StompTemporaryPrefixTest.java | 335 ++++++++++++++++++ 10 files changed, 496 insertions(+), 26 deletions(-) create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java index c8923249405..003e9666212 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java @@ -82,6 +82,8 @@ public static String getMirrorAddress(String connectionName) { private final Map prefixes = new HashMap<>(); + private final Map temporaryPrefixes = new HashMap<>(); + /** * minLargeMessageSize determines when a message should be considered as large. minLargeMessageSize = -1 basically * disables large message control over AMQP. @@ -391,11 +393,32 @@ public void setMulticastPrefix(String multicastPrefix) { } } + @Override + public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { + for (String prefix : temporaryQueuePrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + } + } + + @Override + public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { + for (String prefix : temporaryTopicPrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + } + } + @Override public Map getPrefixes() { return prefixes; } + @Override + public Map getTemporaryPrefixes() { + return temporaryPrefixes; + } + @Override public AMQPRoutingHandler getRoutingHandler() { return routingHandler; diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java index 7b4375837ed..d89fd0f3c4d 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java +++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java @@ -163,6 +163,8 @@ public OpenWireProtocolManager setOpenwireMaxPacketChunkSize(int openwireMaxPack private final Map prefixes = new HashMap<>(); + private final Map temporaryPrefixes = new HashMap<>(); + private final List incomingInterceptors = new ArrayList<>(); private final List outgoingInterceptors = new ArrayList<>(); @@ -686,11 +688,32 @@ public void setMulticastPrefix(String multicastPrefix) { } } + @Override + public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { + for (String prefix : temporaryQueuePrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + } + } + + @Override + public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { + for (String prefix : temporaryTopicPrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + } + } + @Override public Map getPrefixes() { return prefixes; } + @Override + public Map getTemporaryPrefixes() { + return temporaryPrefixes; + } + @Override public void setSecurityDomain(String securityDomain) { this.securityDomain = securityDomain; diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java index 3f7636226db..7ee2f94ffa4 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java @@ -549,11 +549,14 @@ StompPostReceiptFunction subscribe(String destination, String durableSubscriptionName, boolean noLocal, RoutingType subscriptionType, - Integer consumerWindowSize) throws ActiveMQStompException { + Integer consumerWindowSize, + RoutingType temporaryRoutingType) throws ActiveMQStompException { validateSelector(selector); - autoCreateDestinationIfPossible(destination, subscriptionType); - checkDestination(destination); - checkRoutingSemantics(destination, subscriptionType); + if (temporaryRoutingType == null) { + autoCreateDestinationIfPossible(destination, subscriptionType); + checkDestination(destination); + checkRoutingSemantics(destination, subscriptionType); + } if (noLocal) { String noLocalFilter = "(" + CONNECTION_ID_PROPERTY_NAME_STRING + " <> '" + getID().toString() + "' OR " + CONNECTION_ID_PROPERTY_NAME_STRING + " IS NULL)"; if (selector == null) { @@ -578,7 +581,7 @@ StompPostReceiptFunction subscribe(String destination, } try { - return manager.subscribe(this, subscriptionID, durableSubscriptionName, destination, selector, ack, noLocal, consumerWindowSize); + return manager.subscribe(this, subscriptionID, durableSubscriptionName, destination, selector, ack, noLocal, consumerWindowSize, temporaryRoutingType); } catch (ActiveMQStompException e) { throw e; } catch (Exception e) { diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java index 83c9710b3ee..bb41f7dce58 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java @@ -29,6 +29,7 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.ActiveMQExceptionType; import org.apache.activemq.artemis.api.core.BaseInterceptor; +import org.apache.activemq.artemis.api.core.RoutingType; import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.client.ActiveMQClient; import org.apache.activemq.artemis.core.io.IOCallback; @@ -358,14 +359,15 @@ public StompPostReceiptFunction subscribe(StompConnection connection, String selector, String ack, boolean noLocal, - Integer consumerWindowSize) throws Exception { + Integer consumerWindowSize, + RoutingType temporaryRoutingType) throws Exception { StompSession stompSession = getSession(connection); if (stompSession.containsSubscription(subscriptionID)) { throw new ActiveMQStompException(connection, "There already is a subscription for: " + subscriptionID + ". Either use unique subscription IDs or do not create multiple subscriptions for the same destination"); } long consumerID = server.getStorageManager().generateID(); - return stompSession.addSubscription(consumerID, subscriptionID, connection.getClientID(), durableSubscriptionName, destination, selector, ack, noLocal, consumerWindowSize); + return stompSession.addSubscription(consumerID, subscriptionID, connection.getClientID(), durableSubscriptionName, destination, selector, ack, noLocal, consumerWindowSize, temporaryRoutingType); } public void unsubscribe(StompConnection connection, diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompSession.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompSession.java index 9444ee58eba..4d1ce02a243 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompSession.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompSession.java @@ -313,7 +313,8 @@ public StompPostReceiptFunction addSubscription(long consumerID, String selector, String ack, boolean noLocal, - Integer consumerWindowSize) throws Exception { + Integer consumerWindowSize, + RoutingType temporaryRoutingType) throws Exception { SimpleString address = SimpleString.of(destination); SimpleString queueName = SimpleString.of(destination); SimpleString selectorSimple = SimpleString.of(selector); @@ -327,26 +328,42 @@ public StompPostReceiptFunction addSubscription(long consumerID, finalConsumerWindowSize = ConfigurationHelper.getIntProperty(TransportConstants.STOMP_CONSUMER_WINDOW_SIZE, ConfigurationHelper.getIntProperty(TransportConstants.STOMP_CONSUMERS_CREDIT, TransportConstants.STOMP_DEFAULT_CONSUMER_WINDOW_SIZE, connection.getAcceptorUsed().getConfiguration()), connection.getAcceptorUsed().getConfiguration()); } - Set routingTypes = manager.getServer().getAddressInfo(getCoreSession().removePrefix(address)).getRoutingTypes(); - boolean multicast = routingTypes.size() == 1 && routingTypes.contains(RoutingType.MULTICAST); - // if the destination is FQQN then the queue will have already been created - if (multicast && !CompositeAddress.isFullyQualified(destination)) { - // subscribes to a topic - if (durableSubscriptionName != null) { - if (clientID == null) { - throw BUNDLE.missingClientID(); + boolean multicast; + if (temporaryRoutingType != null) { + multicast = temporaryRoutingType == RoutingType.MULTICAST; + if (multicast) { + queueName = UUIDGenerator.getInstance().generateSimpleStringUUID(); + session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setRoutingType(RoutingType.MULTICAST).setFilterString(selectorSimple).setDurable(false).setTemporary(true)); + } else { + try { + session.createQueue(QueueConfiguration.of(address).setAddress(address).setRoutingType(RoutingType.ANYCAST).setFilterString(selectorSimple).setDurable(false).setTemporary(true)); + } catch (ActiveMQQueueExistsException e) { + // queue may already exist if a sender created it first } - queueName = SimpleString.of(clientID + "." + durableSubscriptionName); - if (manager.getServer().locateQueue(queueName) == null) { - try { - session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple)); - } catch (ActiveMQQueueExistsException e) { - // ignore; can be caused by concurrent durable subscribers + queueName = address; + } + } else { + Set routingTypes = manager.getServer().getAddressInfo(getCoreSession().removePrefix(address)).getRoutingTypes(); + multicast = routingTypes.size() == 1 && routingTypes.contains(RoutingType.MULTICAST); + // if the destination is FQQN then the queue will have already been created + if (multicast && !CompositeAddress.isFullyQualified(destination)) { + // subscribes to a topic + if (durableSubscriptionName != null) { + if (clientID == null) { + throw BUNDLE.missingClientID(); } + queueName = SimpleString.of(clientID + "." + durableSubscriptionName); + if (manager.getServer().locateQueue(queueName) == null) { + try { + session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple)); + } catch (ActiveMQQueueExistsException e) { + // ignore; can be caused by concurrent durable subscribers + } + } + } else { + queueName = UUIDGenerator.getInstance().generateSimpleStringUUID(); + session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple).setDurable(false).setTemporary(true)); } - } else { - queueName = UUIDGenerator.getInstance().generateSimpleStringUUID(); - session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple).setDurable(false).setTemporary(true)); } } if (noLocal) { diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java index 1f09cc9c6bb..bff9727c7f8 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java @@ -18,6 +18,7 @@ import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.Map; import java.util.Objects; import java.util.concurrent.ScheduledExecutorService; @@ -263,6 +264,8 @@ public StompFrame onAbort(StompFrame request) { } public StompPostReceiptFunction onSubscribe(StompFrame frame) throws Exception { + String rawDestination = frame.getHeader(Headers.Subscribe.DESTINATION); + RoutingType temporaryRoutingType = getTemporaryRoutingType(rawDestination); String destination = getDestination(frame); String selector = frame.getHeader(Stomp.Headers.Subscribe.SELECTOR); @@ -288,7 +291,19 @@ public StompPostReceiptFunction onSubscribe(StompFrame frame) throws Exception { } else if (frame.hasHeader(Headers.Subscribe.ACTIVEMQ_PREFETCH_SIZE)) { consumerWindowSize = Integer.parseInt(frame.getHeader(Stomp.Headers.Subscribe.ACTIVEMQ_PREFETCH_SIZE)); } - return connection.subscribe(destination, selector, ack, id, durableSubscriptionName, noLocal, routingType, consumerWindowSize); + return connection.subscribe(destination, selector, ack, id, durableSubscriptionName, noLocal, routingType, consumerWindowSize, temporaryRoutingType); + } + + private RoutingType getTemporaryRoutingType(String rawDestination) { + if (rawDestination != null) { + SimpleString dest = SimpleString.of(rawDestination); + for (Map.Entry entry : connection.getManager().getTemporaryPrefixes().entrySet()) { + if (dest.startsWith(entry.getKey())) { + return entry.getValue(); + } + } + } + return null; } public String getDestination(StompFrame request) throws Exception { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java index c8a17b09a57..4293ade21a6 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java @@ -94,6 +94,8 @@ public class CoreProtocolManager implements ProtocolManager prefixes = new HashMap<>(); + private final Map temporaryPrefixes = new HashMap<>(); + private String securityDomain; private final ActiveMQRoutingHandler routingHandler; @@ -227,11 +229,32 @@ public void setMulticastPrefix(String multicastPrefix) { } } + @Override + public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { + for (String prefix : temporaryQueuePrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + } + } + + @Override + public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { + for (String prefix : temporaryTopicPrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + } + } + @Override public Map getPrefixes() { return prefixes; } + @Override + public Map getTemporaryPrefixes() { + return temporaryPrefixes; + } + @Override public void setSecurityDomain(String securityDomain) { this.securityDomain = securityDomain; diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java index 4b10b90a599..0171a19a2ea 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java @@ -33,6 +33,8 @@ public abstract class AbstractProtocolManager, C private final Map prefixes = new HashMap<>(); + private final Map temporaryPrefixes = new HashMap<>(); + private String securityDomain; protected String invokeInterceptors(final List interceptors, final P message, final C connection) { @@ -65,11 +67,32 @@ public void setMulticastPrefix(String multicastPrefix) { } } + @Override + public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { + for (String prefix : temporaryQueuePrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); + } + } + + @Override + public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { + for (String prefix : temporaryTopicPrefix.split(",")) { + prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); + } + } + @Override public Map getPrefixes() { return prefixes; } + @Override + public Map getTemporaryPrefixes() { + return temporaryPrefixes; + } + @Override public String getSecurityDomain() { return securityDomain; diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java index 60b94c639bd..465bb594746 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java @@ -75,8 +75,14 @@ default void removeHandler(String name) { void setMulticastPrefix(String multicastPrefix); + void setTemporaryQueuePrefix(String temporaryQueuePrefix); + + void setTemporaryTopicPrefix(String temporaryTopicPrefix); + Map getPrefixes(); + Map getTemporaryPrefixes(); + void setSecurityDomain(String securityDomain); String getSecurityDomain(); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java new file mode 100644 index 00000000000..b9f73952b1e --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java @@ -0,0 +1,335 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.stomp; + +import java.net.URI; +import java.util.Collection; +import java.util.UUID; + +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.postoffice.Binding; +import org.apache.activemq.artemis.core.postoffice.QueueBinding; +import org.apache.activemq.artemis.core.protocol.stomp.Stomp; +import org.apache.activemq.artemis.core.protocol.stomp.StompProtocolManagerFactory; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.impl.AddressInfo; +import org.apache.activemq.artemis.tests.integration.stomp.util.ClientStompFrame; +import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnection; +import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnectionFactory; +import org.apache.activemq.artemis.tests.util.Wait; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Integration tests for the temporaryQueuePrefix and temporaryTopicPrefix acceptor parameters. + * + * These prefixes allow STOMP clients to subscribe to destinations using a well-known prefix + * (e.g. /temp-queue/ or /temp-topic/) to create temporary resources that are automatically + * deleted when the client disconnects. This mirrors the behaviour of ActiveMQ 5.x. + */ +public class StompTemporaryPrefixTest extends StompTestBase { + + private static final String TEMP_QUEUE_PREFIX = "/temp-queue/"; + private static final String TEMP_TOPIC_PREFIX = "/temp-topic/"; + private static final int TEST_PORT = 61614; + + public StompTemporaryPrefixTest() { + super("tcp+v10.stomp"); + } + + @Override + protected ActiveMQServer createServer() throws Exception { + ActiveMQServer server = super.createServer(); + server.getConfiguration().setAddressQueueScanPeriod(100); + return server; + } + + // --- temporary queue (ANYCAST) tests --- + + @Test + public void testTemporaryQueuePrefixSubscribeCreatesTemporaryAnycastQueue() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String address = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_QUEUE_PREFIX + address) + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + + org.apache.activemq.artemis.core.server.Queue queue = server.locateQueue(SimpleString.of(address)); + assertNotNull(queue, "Subscribing with temporaryQueuePrefix should create an ANYCAST queue"); + assertTrue(queue.isTemporary(), "Queue created by temporaryQueuePrefix should be temporary"); + assertEquals(RoutingType.ANYCAST, queue.getRoutingType()); + + conn.disconnect(); + } + + @Test + public void testTemporaryQueueDeletedOnDisconnect() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String address = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_QUEUE_PREFIX + address) + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + assertNotNull(server.locateQueue(SimpleString.of(address))); + + conn.disconnect(); + + Wait.assertTrue("Temporary ANYCAST queue should be deleted after client disconnects", + () -> server.locateQueue(SimpleString.of(address)) == null); + } + + @Test + public void testSendReceiveViaTemporaryQueue() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String address = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + // Subscribe first so the temp queue exists before sending + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_QUEUE_PREFIX + address) + .addHeader(Stomp.Headers.Subscribe.ID, "sub-1") + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + + send(conn, TEMP_QUEUE_PREFIX + address, null, "Hello Temp Queue", true); + + frame = conn.receiveFrame(5000); + assertNotNull(frame, "Should have received a message on the temporary queue"); + assertEquals(Stomp.Responses.MESSAGE, frame.getCommand()); + assertEquals("Hello Temp Queue", frame.getBody()); + + conn.disconnect(); + } + + // --- temporary topic (MULTICAST) tests --- + + @Test + public void testTemporaryTopicPrefixSubscribeCreatesTemporaryMulticastQueue() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String address = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_TOPIC_PREFIX + address) + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + + AddressInfo addressInfo = server.getAddressInfo(SimpleString.of(address)); + assertNotNull(addressInfo, "Subscribing with temporaryTopicPrefix should create an address"); + assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST), + "Address created by temporaryTopicPrefix should support MULTICAST routing"); + + // A temporary MULTICAST queue (with a UUID name) should be bound to the address + Collection bindings = server.getPostOffice().getDirectBindings(SimpleString.of(address)); + long tempMulticastQueueCount = bindings.stream() + .filter(b -> b instanceof QueueBinding) + .map(b -> ((QueueBinding) b).getQueue()) + .filter(q -> q.isTemporary() && q.getRoutingType() == RoutingType.MULTICAST) + .count(); + assertEquals(1, tempMulticastQueueCount, + "Exactly one temporary MULTICAST queue should be bound to the address"); + + conn.disconnect(); + } + + @Test + public void testTemporaryTopicDeletedOnDisconnect() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String address = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_TOPIC_PREFIX + address) + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + assertNotNull(server.getAddressInfo(SimpleString.of(address))); + + conn.disconnect(); + + Wait.assertTrue("Temporary MULTICAST address should be deleted after client disconnects", + () -> server.getAddressInfo(SimpleString.of(address)) == null); + } + + @Test + public void testSendReceiveViaTemporaryTopic() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String address = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + // Subscribe first so the temp queue exists before sending + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_TOPIC_PREFIX + address) + .addHeader(Stomp.Headers.Subscribe.ID, "sub-1") + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + + send(conn, TEMP_TOPIC_PREFIX + address, null, "Hello Temp Topic", true); + + frame = conn.receiveFrame(5000); + assertNotNull(frame, "Should have received a message on the temporary topic"); + assertEquals(Stomp.Responses.MESSAGE, frame.getCommand()); + assertEquals("Hello Temp Topic", frame.getBody()); + + conn.disconnect(); + } + + // --- combined prefix tests --- + + @Test + public void testBothTemporaryPrefixesOnSameAcceptor() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + String queueAddress = UUID.randomUUID().toString(); + String topicAddress = UUID.randomUUID().toString(); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX + + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + // Subscribe to a temp-queue destination + String receiptId1 = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_QUEUE_PREFIX + queueAddress) + .addHeader(Stomp.Headers.Subscribe.ID, "sub-queue") + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId1); + frame = conn.sendFrame(frame); + assertEquals(receiptId1, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + + org.apache.activemq.artemis.core.server.Queue queue = server.locateQueue(SimpleString.of(queueAddress)); + assertNotNull(queue, "Temp ANYCAST queue should be created"); + assertTrue(queue.isTemporary()); + assertEquals(RoutingType.ANYCAST, queue.getRoutingType()); + + // Subscribe to a temp-topic destination + String receiptId2 = UUID.randomUUID().toString(); + frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, TEMP_TOPIC_PREFIX + topicAddress) + .addHeader(Stomp.Headers.Subscribe.ID, "sub-topic") + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId2); + frame = conn.sendFrame(frame); + assertEquals(receiptId2, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); + + AddressInfo addressInfo = server.getAddressInfo(SimpleString.of(topicAddress)); + assertNotNull(addressInfo, "Temp MULTICAST address should be created"); + assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST)); + + conn.disconnect(); + + Wait.assertTrue("Temp ANYCAST queue should be deleted after disconnect", + () -> server.locateQueue(SimpleString.of(queueAddress)) == null); + Wait.assertTrue("Temp MULTICAST address should be deleted after disconnect", + () -> server.getAddressInfo(SimpleString.of(topicAddress)) == null); + } + + @Test + public void testNonTempSubscriptionStillWorksWithTemporaryPrefixConfigured() throws Exception { + URI uri = createStompClientUri(scheme, hostname, TEST_PORT); + + server.getRemotingService().createAcceptor("test", + "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX + + "&anycastPrefix=/queue/").start(); + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); + conn.connect(defUser, defPass); + + // Subscribe to the regular pre-existing anycast queue using the anycast prefix + String receiptId = UUID.randomUUID().toString(); + ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE) + .addHeader(Stomp.Headers.Subscribe.DESTINATION, "/queue/" + getQueueName()) + .addHeader(Stomp.Headers.Subscribe.ID, "sub-regular") + .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); + frame = conn.sendFrame(frame); + assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID), + "Regular subscription with anycastPrefix should still work alongside temporaryQueuePrefix"); + + send(conn, "/queue/" + getQueueName(), null, "Hello Regular", true); + + frame = conn.receiveFrame(5000); + assertNotNull(frame); + assertEquals(Stomp.Responses.MESSAGE, frame.getCommand()); + assertEquals("Hello Regular", frame.getBody()); + + // Verify the regular queue was NOT created as temporary + org.apache.activemq.artemis.core.server.Queue queue = server.locateQueue(SimpleString.of(getQueueName())); + assertNotNull(queue); + assertFalse(queue.isTemporary(), "Regular queue should not be temporary"); + + conn.disconnect(); + } +} From 08975794b71adaf7d256e8770719ce80f77ae56c Mon Sep 17 00:00:00 2001 From: Anmol Saxena Date: Sun, 26 Apr 2026 07:09:45 +0530 Subject: [PATCH 2/3] ARTEMIS-3164 Rename temporaryQueuePrefix/temporaryTopicPrefix to temporaryAnycastPrefix/temporaryMulticastPrefix. --- .../amqp/broker/ProtonProtocolManager.java | 8 ++--- .../openwire/OpenWireProtocolManager.java | 8 ++--- .../core/impl/CoreProtocolManager.java | 8 ++--- .../protocol/AbstractProtocolManager.java | 8 ++--- .../spi/core/protocol/ProtocolManager.java | 4 +-- .../stomp/StompTemporaryPrefixTest.java | 30 +++++++++---------- 6 files changed, 33 insertions(+), 33 deletions(-) diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java index 003e9666212..73dd3c57f6f 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java @@ -394,16 +394,16 @@ public void setMulticastPrefix(String multicastPrefix) { } @Override - public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { - for (String prefix : temporaryQueuePrefix.split(",")) { + public void setTemporaryAnycastPrefix(String temporaryAnycastPrefix) { + for (String prefix : temporaryAnycastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); } } @Override - public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { - for (String prefix : temporaryTopicPrefix.split(",")) { + public void setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); } diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java index d89fd0f3c4d..1299805a7d4 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java +++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java @@ -689,16 +689,16 @@ public void setMulticastPrefix(String multicastPrefix) { } @Override - public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { - for (String prefix : temporaryQueuePrefix.split(",")) { + public void setTemporaryAnycastPrefix(String temporaryAnycastPrefix) { + for (String prefix : temporaryAnycastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); } } @Override - public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { - for (String prefix : temporaryTopicPrefix.split(",")) { + public void setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java index 4293ade21a6..6e86e76cab6 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java @@ -230,16 +230,16 @@ public void setMulticastPrefix(String multicastPrefix) { } @Override - public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { - for (String prefix : temporaryQueuePrefix.split(",")) { + public void setTemporaryAnycastPrefix(String temporaryAnycastPrefix) { + for (String prefix : temporaryAnycastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); } } @Override - public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { - for (String prefix : temporaryTopicPrefix.split(",")) { + public void setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java index 0171a19a2ea..689f5e44a98 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java @@ -68,16 +68,16 @@ public void setMulticastPrefix(String multicastPrefix) { } @Override - public void setTemporaryQueuePrefix(String temporaryQueuePrefix) { - for (String prefix : temporaryQueuePrefix.split(",")) { + public void setTemporaryAnycastPrefix(String temporaryAnycastPrefix) { + for (String prefix : temporaryAnycastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.ANYCAST); } } @Override - public void setTemporaryTopicPrefix(String temporaryTopicPrefix) { - for (String prefix : temporaryTopicPrefix.split(",")) { + public void setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.split(",")) { prefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); temporaryPrefixes.put(SimpleString.of(prefix), RoutingType.MULTICAST); } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java index 465bb594746..f1e774d42b8 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java @@ -75,9 +75,9 @@ default void removeHandler(String name) { void setMulticastPrefix(String multicastPrefix); - void setTemporaryQueuePrefix(String temporaryQueuePrefix); + void setTemporaryAnycastPrefix(String temporaryAnycastPrefix); - void setTemporaryTopicPrefix(String temporaryTopicPrefix); + void setTemporaryMulticastPrefix(String temporaryMulticastPrefix); Map getPrefixes(); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java index b9f73952b1e..466d67ccdaf 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTemporaryPrefixTest.java @@ -40,7 +40,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; /** - * Integration tests for the temporaryQueuePrefix and temporaryTopicPrefix acceptor parameters. + * Integration tests for the temporaryAnycastPrefix and temporaryMulticastPrefix acceptor parameters. * * These prefixes allow STOMP clients to subscribe to destinations using a well-known prefix * (e.g. /temp-queue/ or /temp-topic/) to create temporary resources that are automatically @@ -72,7 +72,7 @@ public void testTemporaryQueuePrefixSubscribeCreatesTemporaryAnycastQueue() thro server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX).start(); + + "&temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -85,8 +85,8 @@ public void testTemporaryQueuePrefixSubscribeCreatesTemporaryAnycastQueue() thro assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); org.apache.activemq.artemis.core.server.Queue queue = server.locateQueue(SimpleString.of(address)); - assertNotNull(queue, "Subscribing with temporaryQueuePrefix should create an ANYCAST queue"); - assertTrue(queue.isTemporary(), "Queue created by temporaryQueuePrefix should be temporary"); + assertNotNull(queue, "Subscribing with temporaryAnycastPrefix should create an ANYCAST queue"); + assertTrue(queue.isTemporary(), "Queue created by temporaryAnycastPrefix should be temporary"); assertEquals(RoutingType.ANYCAST, queue.getRoutingType()); conn.disconnect(); @@ -99,7 +99,7 @@ public void testTemporaryQueueDeletedOnDisconnect() throws Exception { server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX).start(); + + "&temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -125,7 +125,7 @@ public void testSendReceiveViaTemporaryQueue() throws Exception { server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX).start(); + + "&temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -158,7 +158,7 @@ public void testTemporaryTopicPrefixSubscribeCreatesTemporaryMulticastQueue() th server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + "&temporaryMulticastPrefix=" + TEMP_TOPIC_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -171,9 +171,9 @@ public void testTemporaryTopicPrefixSubscribeCreatesTemporaryMulticastQueue() th assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID)); AddressInfo addressInfo = server.getAddressInfo(SimpleString.of(address)); - assertNotNull(addressInfo, "Subscribing with temporaryTopicPrefix should create an address"); + assertNotNull(addressInfo, "Subscribing with temporaryMulticastPrefix should create an address"); assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST), - "Address created by temporaryTopicPrefix should support MULTICAST routing"); + "Address created by temporaryMulticastPrefix should support MULTICAST routing"); // A temporary MULTICAST queue (with a UUID name) should be bound to the address Collection bindings = server.getPostOffice().getDirectBindings(SimpleString.of(address)); @@ -195,7 +195,7 @@ public void testTemporaryTopicDeletedOnDisconnect() throws Exception { server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + "&temporaryMulticastPrefix=" + TEMP_TOPIC_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -221,7 +221,7 @@ public void testSendReceiveViaTemporaryTopic() throws Exception { server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + "&temporaryMulticastPrefix=" + TEMP_TOPIC_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -255,8 +255,8 @@ public void testBothTemporaryPrefixesOnSameAcceptor() throws Exception { server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX - + "&temporaryTopicPrefix=" + TEMP_TOPIC_PREFIX).start(); + + "&temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX + + "&temporaryMulticastPrefix=" + TEMP_TOPIC_PREFIX).start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); @@ -302,7 +302,7 @@ public void testNonTempSubscriptionStillWorksWithTemporaryPrefixConfigured() thr server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + TEST_PORT + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME - + "&temporaryQueuePrefix=" + TEMP_QUEUE_PREFIX + + "&temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX + "&anycastPrefix=/queue/").start(); StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri); @@ -316,7 +316,7 @@ public void testNonTempSubscriptionStillWorksWithTemporaryPrefixConfigured() thr .addHeader(Stomp.Headers.RECEIPT_REQUESTED, receiptId); frame = conn.sendFrame(frame); assertEquals(receiptId, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID), - "Regular subscription with anycastPrefix should still work alongside temporaryQueuePrefix"); + "Regular subscription with anycastPrefix should still work alongside temporaryAnycastPrefix"); send(conn, "/queue/" + getQueueName(), null, "Hello Regular", true); From 18bac5dd4ecc85432bf386fb86dc3f3918f004ab Mon Sep 17 00:00:00 2001 From: Anmol Saxena Date: Tue, 21 Jul 2026 22:13:44 +0530 Subject: [PATCH 3/3] ARTEMIS-3164 Move temporary prefix support from STOMP into the core session. --- .../amqp/broker/AMQPSessionCallback.java | 4 +- .../protocol/mqtt/MQTTConnectionManager.java | 1 + .../protocol/openwire/OpenWireConnection.java | 2 +- .../protocol/openwire/amq/AMQSession.java | 2 +- .../openwire/amq/OpenWireConnectionTest.java | 2 +- .../amq/OpenWireProtocolManagerTest.java | 2 +- .../core/protocol/stomp/StompConnection.java | 26 +-- .../protocol/stomp/StompProtocolManager.java | 8 +- .../core/protocol/stomp/StompSession.java | 53 ++---- .../stomp/VersionedStompFrameHandler.java | 28 +-- .../core/management/impl/AbstractControl.java | 2 +- .../core/impl/ActiveMQPacketHandler.java | 2 +- .../artemis/core/server/ActiveMQServer.java | 2 + .../artemis/core/server/ServerSession.java | 11 ++ .../core/server/impl/ActiveMQServerImpl.java | 9 +- .../core/server/impl/ServerSessionImpl.java | 49 ++++- .../server/impl/ServerSessionImplTest.java | 2 +- .../integration/client/HangConsumerTest.java | 3 +- .../client/TemporaryPrefixTest.java | 171 ++++++++++++++++++ .../consumer/OrphanedConsumerDefenseTest.java | 2 +- 20 files changed, 293 insertions(+), 88 deletions(-) create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/TemporaryPrefixTest.java diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java index 35a8eb28cc7..e5b70a385dc 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java @@ -210,14 +210,14 @@ public void init(AMQPSessionContext protonSession, SASLResult saslResult) throws false, // boolean autoCommitAcks, false, // boolean preAcknowledge, true, //boolean xa, - null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain(), false); + null, this, true, operationContext, manager.getPrefixes(), manager.getTemporaryPrefixes(), manager.getSecurityDomain(), false); } else { serverSession = manager.getServer().createSession(name, connection.getUser(), connection.getPassword(), ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, protonSPI.getProtonConnectionDelegate(), // RemotingConnection remotingConnection, false, // boolean autoCommitSends false, // boolean autoCommitAcks, false, // boolean preAcknowledge, true, //boolean xa, - null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain(), connection.getValidatedUser(), false); + null, this, true, operationContext, manager.getPrefixes(), manager.getTemporaryPrefixes(), manager.getSecurityDomain(), connection.getValidatedUser(), false); } } diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java index 1a4031262d4..60e26ede432 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java @@ -168,6 +168,7 @@ ServerSessionImpl createServerSession(String username, String password, String v MQTTUtil.SESSION_AUTO_CREATE_QUEUE, session.getSessionContext(), session.getProtocolManager().getPrefixes(), + session.getProtocolManager().getTemporaryPrefixes(), session.getProtocolManager().getSecurityDomain(), validatedUser, false); diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java index 8f7105ae11a..7ace1dad9f0 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java +++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java @@ -883,7 +883,7 @@ private static int getSize(Command command) { } private void createInternalSession(ConnectionInfo info) throws Exception { - internalSession = server.createSession(UUIDGenerator.getInstance().generateStringUUID(), context.getUserName(), info.getPassword(), -1, this, true, false, false, false, null, null, true, operationContext, protocolManager.getPrefixes(), protocolManager.getSecurityDomain(), validatedUser, false); + internalSession = server.createSession(UUIDGenerator.getInstance().generateStringUUID(), context.getUserName(), info.getPassword(), -1, this, true, false, false, false, null, null, true, operationContext, protocolManager.getPrefixes(), protocolManager.getTemporaryPrefixes(), protocolManager.getSecurityDomain(), validatedUser, false); } /** diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java index ad035b03031..f404be6e47d 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java +++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java @@ -134,7 +134,7 @@ public void initialize() { // now try { - coreSession = server.createSession(name, username, password, minLargeMessageSize, connection, true, false, false, false, null, this, true, connection.getOperationContext(), protocolManager.getPrefixes(), protocolManager.getSecurityDomain(), connection.getValidatedUser(), false); + coreSession = server.createSession(name, username, password, minLargeMessageSize, connection, true, false, false, false, null, this, true, connection.getOperationContext(), protocolManager.getPrefixes(), protocolManager.getTemporaryPrefixes(), protocolManager.getSecurityDomain(), connection.getValidatedUser(), false); } catch (Exception e) { logger.error("error init session", e); } diff --git a/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireConnectionTest.java b/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireConnectionTest.java index 6868fca3777..90b3305028b 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireConnectionTest.java +++ b/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireConnectionTest.java @@ -71,7 +71,7 @@ public void testActorStateVisibility() throws Exception { ServerSession serverSession = Mockito.mock(ServerSession.class); Mockito.when(serverSession.getName()).thenReturn("session"); Mockito.doReturn(serverSession).when(server).createSession(Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), - Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.anyBoolean()); + Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.anyBoolean()); OpenWireProtocolManager openWireProtocolManager = new OpenWireProtocolManager(null, server, null, null); openWireProtocolManager.setSecurityDomain("securityDomain"); diff --git a/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireProtocolManagerTest.java b/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireProtocolManagerTest.java index 6609f03c122..bc101b5f2cb 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireProtocolManagerTest.java +++ b/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireProtocolManagerTest.java @@ -59,7 +59,7 @@ public void testNullPrimaryOnNodeUp() throws Exception { Mockito.when(securityStore.authenticate(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(null); ServerSession serverSession = Mockito.mock(ServerSession.class); Mockito.when(serverSession.getName()).thenReturn("session"); - Mockito.doReturn(serverSession).when(server).createSession(Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.anyBoolean()); + Mockito.doReturn(serverSession).when(server).createSession(Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.anyBoolean()); OpenWireProtocolManager openWireProtocolManager = new OpenWireProtocolManager(null, server, null, null); openWireProtocolManager.setSecurityDomain("securityDomain"); diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java index 50e08d4f289..411d11118bf 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java @@ -486,9 +486,10 @@ StompPostReceiptFunction subscribe(String destination, String durableSubscriptionName, boolean noLocal, RoutingType subscriptionType, - Integer consumerWindowSize) throws ActiveMQStompException { + Integer consumerWindowSize, + boolean temporary) throws ActiveMQStompException { validateSelector(selector); - checkAutoCreate(destination, subscriptionType); + checkAutoCreate(destination, subscriptionType, temporary); String subscriptionID = getSubscriptionID(destination, id); try { @@ -507,11 +508,15 @@ StompPostReceiptFunction subscribe(String destination, } } - protected void checkAutoCreate(String destination, RoutingType subscriptionType) throws ActiveMQStompException { + protected void checkAutoCreate(String destination, RoutingType subscriptionType, boolean temporary) throws ActiveMQStompException { AutoCreateResult autoCreateResult; try { RoutingType routingType = getSubscriptionRoutingType(destination, subscriptionType); - autoCreateResult = getSession().getCoreSession().checkAutoCreate(QueueConfiguration.of(destination).setRoutingType(routingType)); + QueueConfiguration queueConfiguration = QueueConfiguration.of(destination).setRoutingType(routingType); + if (temporary) { + queueConfiguration.setTemporary(true).setDurable(false); + } + autoCreateResult = getSession().getCoreSession().checkAutoCreate(queueConfiguration); } catch (Exception e) { logger.debug("Exception while auto-creating destination", e); throw new ActiveMQStompException(e.getMessage(), e).setHandler(frameHandler); @@ -530,11 +535,6 @@ private RoutingType getSubscriptionRoutingType(String destination, RoutingType s } private String getSelector(String selector, boolean noLocal) { - if (temporaryRoutingType == null) { - autoCreateDestinationIfPossible(destination, subscriptionType); - checkDestination(destination); - checkRoutingSemantics(destination, subscriptionType); - } if (noLocal) { String noLocalFilter = "(" + CONNECTION_ID_PROPERTY_NAME_STRING + " <> '" + getID().toString() + "' OR " + CONNECTION_ID_PROPERTY_NAME_STRING + " IS NULL)"; if (selector == null) { @@ -556,14 +556,6 @@ private String getSubscriptionID(String destination, String id) throws ActiveMQS } subscriptionID = "subscription/" + destination; } - - try { - return manager.subscribe(this, subscriptionID, durableSubscriptionName, destination, selector, ack, noLocal, consumerWindowSize, temporaryRoutingType); - } catch (ActiveMQStompException e) { - throw e; - } catch (Exception e) { - throw BUNDLE.errorCreatingSubscription(subscriptionID, e).setHandler(frameHandler); - } return subscriptionID; } diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java index bb41f7dce58..3ee60d13ad4 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java @@ -29,7 +29,6 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.ActiveMQExceptionType; import org.apache.activemq.artemis.api.core.BaseInterceptor; -import org.apache.activemq.artemis.api.core.RoutingType; import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.client.ActiveMQClient; import org.apache.activemq.artemis.core.io.IOCallback; @@ -246,7 +245,7 @@ private StompSession internalGetSession(StompConnection connection, Map routingTypes = manager.getServer().getAddressInfo(getCoreSession().removePrefix(address)).getRoutingTypes(); + boolean multicast = routingTypes.size() == 1 && routingTypes.contains(RoutingType.MULTICAST); + // if the destination is FQQN then the queue will have already been created + if (multicast && !CompositeAddress.isFullyQualified(destination)) { + // subscribes to a topic + if (durableSubscriptionName != null) { + if (clientID == null) { + throw BUNDLE.missingClientID(); } - queueName = address; - } - } else { - Set routingTypes = manager.getServer().getAddressInfo(getCoreSession().removePrefix(address)).getRoutingTypes(); - multicast = routingTypes.size() == 1 && routingTypes.contains(RoutingType.MULTICAST); - // if the destination is FQQN then the queue will have already been created - if (multicast && !CompositeAddress.isFullyQualified(destination)) { - // subscribes to a topic - if (durableSubscriptionName != null) { - if (clientID == null) { - throw BUNDLE.missingClientID(); + queueName = SimpleString.of(clientID + "." + durableSubscriptionName); + if (manager.getServer().locateQueue(queueName) == null) { + try { + session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple)); + } catch (ActiveMQQueueExistsException e) { + // ignore; can be caused by concurrent durable subscribers } - queueName = SimpleString.of(clientID + "." + durableSubscriptionName); - if (manager.getServer().locateQueue(queueName) == null) { - try { - session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple)); - } catch (ActiveMQQueueExistsException e) { - // ignore; can be caused by concurrent durable subscribers - } - } - } else { - queueName = UUIDGenerator.getInstance().generateSimpleStringUUID(); - session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple).setDurable(false).setTemporary(true)); } + } else { + queueName = UUIDGenerator.getInstance().generateSimpleStringUUID(); + session.createQueue(QueueConfiguration.of(queueName).setAddress(address).setFilterString(selectorSimple).setDurable(false).setTemporary(true)); } } if (noLocal) { diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java index 19373b6a355..08c92f27876 100644 --- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java +++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java @@ -17,12 +17,12 @@ package org.apache.activemq.artemis.core.protocol.stomp; import java.nio.charset.StandardCharsets; -import java.util.Map; import java.util.Objects; import java.util.concurrent.ScheduledExecutorService; import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.ActiveMQException; +import org.apache.activemq.artemis.api.core.ActiveMQSecurityException; import org.apache.activemq.artemis.api.core.ICoreMessage; import org.apache.activemq.artemis.api.core.Message; import org.apache.activemq.artemis.api.core.RoutingType; @@ -191,7 +191,8 @@ public StompFrame onSend(StompFrame frame) { connection.validate(); String destination = getDestination(frame); RoutingType routingType = getRoutingType(frame.getHeader(Headers.Send.DESTINATION_TYPE), frame.getHeader(Headers.Send.DESTINATION)); - connection.checkAutoCreate(destination, routingType); + boolean temporary = isTemporaryDestination(frame.getHeader(Headers.Send.DESTINATION)); + connection.checkAutoCreate(destination, routingType, temporary); String txID = frame.getHeader(Stomp.Headers.TRANSACTION); long timestamp = System.currentTimeMillis(); @@ -259,8 +260,6 @@ public StompFrame onAbort(StompFrame request) { } public StompPostReceiptFunction onSubscribe(StompFrame frame) throws Exception { - String rawDestination = frame.getHeader(Headers.Subscribe.DESTINATION); - RoutingType temporaryRoutingType = getTemporaryRoutingType(rawDestination); String destination = getDestination(frame); String selector = frame.getHeader(Stomp.Headers.Subscribe.SELECTOR); @@ -274,6 +273,7 @@ public StompPostReceiptFunction onSubscribe(StompFrame frame) throws Exception { } } RoutingType routingType = getRoutingType(frame.getHeader(Headers.Subscribe.SUBSCRIPTION_TYPE), frame.getHeader(Headers.Subscribe.DESTINATION)); + boolean temporary = isTemporaryDestination(frame.getHeader(Headers.Subscribe.DESTINATION)); boolean noLocal = false; if (frame.hasHeader(Stomp.Headers.Subscribe.NO_LOCAL)) { noLocal = Boolean.parseBoolean(frame.getHeader(Stomp.Headers.Subscribe.NO_LOCAL)); @@ -286,19 +286,19 @@ public StompPostReceiptFunction onSubscribe(StompFrame frame) throws Exception { } else if (frame.hasHeader(Headers.Subscribe.ACTIVEMQ_PREFETCH_SIZE)) { consumerWindowSize = Integer.parseInt(frame.getHeader(Stomp.Headers.Subscribe.ACTIVEMQ_PREFETCH_SIZE)); } - return connection.subscribe(destination, selector, ack, id, durableSubscriptionName, noLocal, routingType, consumerWindowSize, temporaryRoutingType); + return connection.subscribe(destination, selector, ack, id, durableSubscriptionName, noLocal, routingType, consumerWindowSize, temporary); } - private RoutingType getTemporaryRoutingType(String rawDestination) { - if (rawDestination != null) { - SimpleString dest = SimpleString.of(rawDestination); - for (Map.Entry entry : connection.getManager().getTemporaryPrefixes().entrySet()) { - if (dest.startsWith(entry.getKey())) { - return entry.getValue(); - } - } + /** + * The address on a SEND/SUBSCRIBE frame may already have had a temporary prefix stripped from it by + * {@link #getDestination(StompFrame)} by the time it reaches most of this class, so this must be checked against + * the raw (un-stripped) destination header, the same way {@link #getRoutingType} is resolved from the raw header. + */ + private boolean isTemporaryDestination(String rawDestination) throws ActiveMQStompException, ActiveMQSecurityException { + if (rawDestination == null) { + return false; } - return null; + return connection.getSession().getCoreSession().getRoutingTypeFromTemporaryPrefix(SimpleString.of(rawDestination)) != null; } public String getDestination(StompFrame request) throws Exception { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AbstractControl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AbstractControl.java index 275c2b4380f..4b4de886242 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AbstractControl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AbstractControl.java @@ -120,7 +120,7 @@ protected String sendMessage(SimpleString address, Integer.MAX_VALUE, fakeConnection, true, true, false, false, address.toString(), fakeConnection.callback, - false, new DummyOperationContext(), Collections.emptyMap(), null, validatedUser, false); + false, new DummyOperationContext(), Collections.emptyMap(), Collections.emptyMap(), null, validatedUser, false); try { CoreMessage message = new CoreMessage(storageManager.generateID(), 50); if (headers != null) { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQPacketHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQPacketHandler.java index 87fc1c00c1c..8df7c2de435 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQPacketHandler.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQPacketHandler.java @@ -191,7 +191,7 @@ private void handleCreateSession(final CreateSessionMessage request) { CoreSessionCallback sessionCallback = new CoreSessionCallback(request.getName(), protocolManager, channel, connection); boolean isLegacyProducer = request.getVersion() < PacketImpl.ARTEMIS_2_28_0_VERSION; - ServerSession session = server.createSession(request.getName(), activeMQPrincipal == null ? request.getUsername() : activeMQPrincipal.getUserName(), activeMQPrincipal == null ? request.getPassword() : activeMQPrincipal.getPassword(), request.getMinLargeMessageSize(), connection, request.isAutoCommitSends(), request.isAutoCommitAcks(), request.isPreAcknowledge(), request.isXA(), request.getDefaultAddress(), sessionCallback, true, sessionOperationContext, routingTypeMap, protocolManager.getSecurityDomain(), validatedUser, isLegacyProducer); + ServerSession session = server.createSession(request.getName(), activeMQPrincipal == null ? request.getUsername() : activeMQPrincipal.getUserName(), activeMQPrincipal == null ? request.getPassword() : activeMQPrincipal.getPassword(), request.getMinLargeMessageSize(), connection, request.isAutoCommitSends(), request.isAutoCommitAcks(), request.isPreAcknowledge(), request.isXA(), request.getDefaultAddress(), sessionCallback, true, sessionOperationContext, routingTypeMap, protocolManager.getTemporaryPrefixes(), protocolManager.getSecurityDomain(), validatedUser, isLegacyProducer); ServerSessionPacketHandler handler = new ServerSessionPacketHandler(server, session, channel); channel.setHandler(handler); sessionCallback.setSessionHandler(handler); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java index ac39f28f041..072a9def821 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java @@ -383,6 +383,7 @@ ServerSession createSession(String name, boolean autoCreateQueues, OperationContext context, Map prefixes, + Map temporaryPrefixes, String securityDomain, String validatedUser, boolean isLegacyProducer) throws Exception; @@ -402,6 +403,7 @@ ServerSession createInternalSession(String name, boolean autoCreateQueues, OperationContext context, Map prefixes, + Map temporaryPrefixes, String securityDomain, boolean isLegacyProducer) throws Exception; diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java index 623494986eb..448699a1961 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java @@ -526,6 +526,17 @@ void createSharedQueue(SimpleString address, RoutingType getRoutingTypeFromPrefix(SimpleString address, RoutingType defaultRoutingType); + /** + * Returns the routing type registered for {@code address} under one of this session's temporary prefixes (e.g. + * {@code temporaryAnycastPrefix}/{@code temporaryMulticastPrefix}), or {@code null} if {@code address} doesn't + * match any of them. A non-null result means the resource being created for this address should be forced + * temporary, regardless of what the client requested. + * + * @param address the address to inspect + * @return the {@code RoutingType} registered for the matching temporary prefix, or {@code null} if none match + */ + RoutingType getRoutingTypeFromTemporaryPrefix(SimpleString address); + /** * Get the canonical (i.e. non-prefixed) address and the corresponding routing-type. * diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java index 4fb63b0b006..db81b3e1de4 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java @@ -1862,6 +1862,7 @@ public ServerSession createSession(final String name, final boolean autoCreateQueues, final OperationContext context, final Map prefixes, + final Map temporaryPrefixes, final String securityDomain, String validatedUser, boolean isLegacyProducer) throws Exception { @@ -1876,7 +1877,7 @@ public ServerSession createSession(final String name, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, autoCreateQueues, prefixes); } - final ServerSessionImpl session = internalCreateSession(name, username, password, validatedUser, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain, isLegacyProducer); + final ServerSessionImpl session = internalCreateSession(name, username, password, validatedUser, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, temporaryPrefixes, securityDomain, isLegacyProducer); return session; } @@ -1904,9 +1905,10 @@ public ServerSession createInternalSession(String name, boolean autoCreateQueues, OperationContext context, Map prefixes, + Map temporaryPrefixes, String securityDomain, boolean isLegacyProducer) throws Exception { - ServerSessionImpl session = internalCreateSession(name, null, null, null, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain, isLegacyProducer); + ServerSessionImpl session = internalCreateSession(name, null, null, null, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, temporaryPrefixes, securityDomain, isLegacyProducer); session.disableSecurity(); return session; } @@ -1982,6 +1984,7 @@ protected ServerSessionImpl internalCreateSession(String name, OperationContext context, boolean autoCreateQueues, Map prefixes, + Map temporaryPrefixes, String securityDomain, boolean isLegacyProducer) throws Exception { @@ -1990,7 +1993,7 @@ protected ServerSessionImpl internalCreateSession(String name, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, autoCreateQueues, context, prefixes)); } - ServerSessionImpl session = new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, this, defaultAddress == null ? null : SimpleString.of(defaultAddress), callback, context, prefixes, securityDomain, isLegacyProducer); + ServerSessionImpl session = new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, this, defaultAddress == null ? null : SimpleString.of(defaultAddress), callback, context, prefixes, temporaryPrefixes, securityDomain, isLegacyProducer); sessions.put(name, session); totalSessionCount.incrementAndGet(); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java index 01764ae8e58..8aa987375e7 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java @@ -215,6 +215,10 @@ public class ServerSessionImpl extends CriticalComponentImpl implements ServerSe private Map prefixes; + private boolean temporaryPrefixEnabled = false; + + private Map temporaryPrefixes; + private Set closeables; private final Executor sessionExecutor; @@ -235,6 +239,7 @@ public ServerSessionImpl(final String name, final SessionCallback callback, final OperationContext context, final Map prefixes, + final Map temporaryPrefixes, final String securityDomain, boolean isLegacyProducer) throws Exception { super(server.getCriticalAnalyzer(), 1); @@ -281,6 +286,11 @@ public ServerSessionImpl(final String name, prefixEnabled = true; } + this.temporaryPrefixes = temporaryPrefixes; + if (this.temporaryPrefixes != null && !this.temporaryPrefixes.isEmpty()) { + temporaryPrefixEnabled = true; + } + this.managementAddress = managementService.getManagementAddress(); this.callback = callback; @@ -777,11 +787,18 @@ public Queue createQueue(QueueConfiguration queueConfiguration) throws Exception queueConfiguration.setInternal(true); } + SimpleString originalAddress = queueConfiguration.getAddress(); + RoutingType temporaryRoutingType = getRoutingTypeFromTemporaryPrefix(originalAddress); + queueConfiguration - .setRoutingType(getRoutingTypeFromPrefix(queueConfiguration.getAddress(), queueConfiguration.getRoutingType())) - .setAddress(removePrefix(queueConfiguration.getAddress())) + .setRoutingType(getRoutingTypeFromPrefix(originalAddress, queueConfiguration.getRoutingType())) + .setAddress(removePrefix(originalAddress)) .setName(removePrefix(queueConfiguration.getName())); + if (temporaryRoutingType != null) { + queueConfiguration.setTemporary(true).setDurable(false); + } + // make sure the user has privileges to create this queue securityCheck(queueConfiguration.getAddress(), queueConfiguration.getName(), queueConfiguration.isDurable() ? CheckType.CREATE_DURABLE_QUEUE : CheckType.CREATE_NON_DURABLE_QUEUE, this); @@ -1008,10 +1025,13 @@ public AddressInfo createAddress(AddressInfo addressInfo, boolean autoCreated) t } AddressInfo art = getAddressAndRoutingType(addressInfo); + if (getRoutingTypeFromTemporaryPrefix(addressInfo.getName()) != null) { + art.setTemporary(true); + } securityCheck(art.getName(), CheckType.CREATE_ADDRESS, this); server.addOrUpdateAddressInfo(art.setAutoCreated(autoCreated)); if (art.isTemporary()) { - handleTempResource(addressInfo.getName(), false); + handleTempResource(art.getName(), false); } return server.getAddressInfo(art.getName()); } @@ -1876,6 +1896,10 @@ public Transaction getCurrentTransaction() { @Override public AutoCreateResult checkAutoCreate(final QueueConfiguration queueConfig) throws Exception { AutoCreateResult result; + RoutingType temporaryRoutingType = getRoutingTypeFromTemporaryPrefix(queueConfig.getAddress()); + if (temporaryRoutingType != null) { + queueConfig.setRoutingType(temporaryRoutingType).setTemporary(true).setDurable(false); + } SimpleString unPrefixedAddress = removePrefix(queueConfig.getAddress()); SimpleString unPrefixedQueue = removePrefix(queueConfig.getName()); AddressSettings addressSettings = server.getAddressSettingsRepository().getMatch(unPrefixedAddress.toString()); @@ -2588,6 +2612,25 @@ public RoutingType getRoutingTypeFromPrefix(SimpleString address, RoutingType de return defaultRoutingType; } + /** + * Returns the routing type registered for {@code address} under one of this session's temporary prefixes + * (e.g. {@code temporaryAnycastPrefix}/{@code temporaryMulticastPrefix}), or {@code null} if {@code address} + * doesn't match any of them. A non-null result means the resource being created for this address should be + * forced temporary, regardless of what the client requested, mirroring how {@link #getRoutingTypeFromPrefix} + * forces the routing type for the plain anycast/multicast prefixes. + */ + @Override + public RoutingType getRoutingTypeFromTemporaryPrefix(SimpleString address) { + if (temporaryPrefixEnabled && address != null) { + for (Map.Entry entry : temporaryPrefixes.entrySet()) { + if (address.startsWith(entry.getKey())) { + return entry.getValue(); + } + } + } + return null; + } + @Override public Pair> getAddressAndRoutingTypes(SimpleString address, EnumSet defaultRoutingTypes) { diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImplTest.java index 757b0249dff..54845d6dcb9 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImplTest.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImplTest.java @@ -50,7 +50,7 @@ public void testCheckAutoCreateModifyExistingAddressInfo() throws Exception { setupMocksForAddressStuff(server, addressInfo); - ServerSessionImpl session = new ServerSessionImpl(null, null, null, null, 0, false, false, false, false, true, mock(RemotingConnection.class), server, null, null, null, null, null, false); + ServerSessionImpl session = new ServerSessionImpl(null, null, null, null, 0, false, false, false, false, true, mock(RemotingConnection.class), server, null, null, null, null, null, null, false); QueueConfiguration queueConfiguration = QueueConfiguration.of(NAME).setRoutingType(RoutingType.MULTICAST); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java index 35bc20b83a7..a4f8dac7917 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java @@ -552,9 +552,10 @@ protected ServerSessionImpl internalCreateSession(String name, OperationContext context, boolean autoCreateQueue, Map prefixes, + Map temporaryPrefixes, String securityDomain, boolean isLegacyProducer) throws Exception { - return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, this, defaultAddress == null ? null : SimpleString.of(defaultAddress), new MyCallback(callback), context, prefixes, securityDomain, isLegacyProducer); + return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, this, defaultAddress == null ? null : SimpleString.of(defaultAddress), new MyCallback(callback), context, prefixes, temporaryPrefixes, securityDomain, isLegacyProducer); } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/TemporaryPrefixTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/TemporaryPrefixTest.java new file mode 100644 index 00000000000..c750ff0f424 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/TemporaryPrefixTest.java @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.client; + +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.client.ClientSession; +import org.apache.activemq.artemis.api.core.client.ClientSessionFactory; +import org.apache.activemq.artemis.api.core.client.ServerLocator; +import org.apache.activemq.artemis.core.client.impl.ServerLocatorImpl; +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.ActiveMQServers; +import org.apache.activemq.artemis.core.server.Queue; +import org.apache.activemq.artemis.core.server.impl.AddressInfo; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.tests.util.Wait; +import org.apache.activemq.artemis.utils.UUIDGenerator; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that {@code temporaryAnycastPrefix} and {@code temporaryMulticastPrefix} are implemented at the Core + * server level (in {@code ServerSessionImpl}) and therefore work for any protocol, exactly like the existing + * {@code anycastPrefix}/{@code multicastPrefix} settings. This drives the feature directly through the Core client + * rather than through any single protocol (e.g. STOMP) to prove the behaviour isn't protocol-specific. + */ +public class TemporaryPrefixTest extends ActiveMQTestBase { + + private static final String TEMP_QUEUE_PREFIX = "temp-queue://"; + private static final String TEMP_TOPIC_PREFIX = "temp-topic://"; + + private static final String TEST_URI = "tcp://localhost:5446"; + + private ActiveMQServer createAndStartServer(String acceptorParams) throws Exception { + Configuration configuration = createBasicConfig(); + configuration.setAddressQueueScanPeriod(100); + configuration.clearAcceptorConfigurations(); + configuration.addAcceptorConfiguration("core", TEST_URI + "?protocols=CORE" + acceptorParams); + ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(configuration, false)); + server.start(); + return server; + } + + private ClientSession createSession() throws Exception { + ServerLocator locator = addServerLocator(ServerLocatorImpl.newLocator(TEST_URI)); + ClientSessionFactory sf = createSessionFactory(locator); + return addClientSession(sf.createSession(false, true, true)); + } + + @Test + public void testTemporaryAnycastPrefixForcesTemporaryQueue() throws Exception { + ActiveMQServer server = createAndStartServer(";temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX); + ClientSession session = createSession(); + String address = UUIDGenerator.getInstance().generateStringUUID(); + + // client asks for a durable, non-temporary queue; Core must override this because of the temp prefix + session.createQueue(QueueConfiguration.of(address).setAddress(TEMP_QUEUE_PREFIX + address).setDurable(true).setTemporary(false)); + + Queue queue = server.locateQueue(SimpleString.of(address)); + assertNotNull(queue, "the temp prefix should have caused the address/queue to be auto-created"); + assertTrue(queue.isTemporary(), "queue created under temporaryAnycastPrefix must be forced temporary"); + assertFalse(queue.isDurable(), "queue created under temporaryAnycastPrefix must be forced non-durable"); + assertEquals(RoutingType.ANYCAST, queue.getRoutingType()); + + session.getSessionFactory().close(); + + Wait.assertTrue("temporary queue should be removed once the session/connection closes", + () -> server.locateQueue(SimpleString.of(address)) == null); + } + + @Test + public void testTemporaryMulticastPrefixForcesTemporaryAddress() throws Exception { + ActiveMQServer server = createAndStartServer(";temporaryMulticastPrefix=" + TEMP_TOPIC_PREFIX); + ClientSession session = createSession(); + String address = UUIDGenerator.getInstance().generateStringUUID(); + + session.createAddress(SimpleString.of(TEMP_TOPIC_PREFIX + address), RoutingType.MULTICAST, false); + + AddressInfo addressInfo = server.getAddressInfo(SimpleString.of(address)); + assertNotNull(addressInfo, "the temp prefix should have caused the address to be auto-created"); + assertTrue(addressInfo.isTemporary(), "address created under temporaryMulticastPrefix must be forced temporary"); + assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST)); + + session.getSessionFactory().close(); + + Wait.assertTrue("temporary address should be removed once the session/connection closes", + () -> server.getAddressInfo(SimpleString.of(address)) == null); + } + + @Test + public void testNonPrefixedAddressStaysNonTemporary() throws Exception { + ActiveMQServer server = createAndStartServer(";temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX); + ClientSession session = createSession(); + String address = UUIDGenerator.getInstance().generateStringUUID(); + + // no temp prefix on this address, so the queue should be created exactly as requested + session.createQueue(QueueConfiguration.of(address).setAddress(address).setDurable(true)); + + Queue queue = server.locateQueue(SimpleString.of(address)); + assertNotNull(queue); + assertFalse(queue.isTemporary(), "queue created without the temp prefix must not be forced temporary"); + assertTrue(queue.isDurable()); + + session.close(); + assertNotNull(server.locateQueue(SimpleString.of(address)), "non-temporary queue must survive session close"); + } + + @Test + public void testBothTemporaryPrefixesOnSameAcceptor() throws Exception { + ActiveMQServer server = createAndStartServer(";temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX + ";temporaryMulticastPrefix=" + TEMP_TOPIC_PREFIX); + ClientSession session = createSession(); + String queueAddress = UUIDGenerator.getInstance().generateStringUUID(); + String topicAddress = UUIDGenerator.getInstance().generateStringUUID(); + + session.createQueue(QueueConfiguration.of(queueAddress).setAddress(TEMP_QUEUE_PREFIX + queueAddress)); + session.createAddress(SimpleString.of(TEMP_TOPIC_PREFIX + topicAddress), RoutingType.MULTICAST, false); + + Queue queue = server.locateQueue(SimpleString.of(queueAddress)); + assertNotNull(queue); + assertTrue(queue.isTemporary()); + assertEquals(RoutingType.ANYCAST, queue.getRoutingType()); + + AddressInfo addressInfo = server.getAddressInfo(SimpleString.of(topicAddress)); + assertNotNull(addressInfo); + assertTrue(addressInfo.isTemporary()); + assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST)); + + session.getSessionFactory().close(); + + Wait.assertTrue(() -> server.locateQueue(SimpleString.of(queueAddress)) == null); + Wait.assertTrue(() -> server.getAddressInfo(SimpleString.of(topicAddress)) == null); + } + + @Test + public void testNoPrefixConfiguredMeansNoOverride() throws Exception { + ActiveMQServer server = createAndStartServer(""); + ClientSession session = createSession(); + String address = UUIDGenerator.getInstance().generateStringUUID(); + + session.createQueue(QueueConfiguration.of(address).setAddress(TEMP_QUEUE_PREFIX + address).setDurable(true)); + + // with no temporaryAnycastPrefix configured, "temp-queue://" is just a normal (unrecognized) address segment + Queue queue = server.locateQueue(SimpleString.of(address)); + assertNotNull(queue); + assertFalse(queue.isTemporary()); + assertTrue(queue.isDurable()); + + session.close(); + assertNotNull(server.locateQueue(SimpleString.of(address))); + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/consumer/OrphanedConsumerDefenseTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/consumer/OrphanedConsumerDefenseTest.java index 9e8134b386e..2d4b9248118 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/consumer/OrphanedConsumerDefenseTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/consumer/OrphanedConsumerDefenseTest.java @@ -121,7 +121,7 @@ public void testDefendDestroyedConnection() throws Exception { remotingConnection.destroy(); ServerSessionImpl session = new ServerSessionImpl(RandomUtil.randomUUIDString(), RandomUtil.randomUUIDString(), RandomUtil.randomUUIDString(), RandomUtil.randomUUIDString(), 1000, true, true, true, true, true, - remotingConnection, mockServer, RandomUtil.randomUUIDSimpleString(), Mockito.mock(SessionCallback.class), Mockito.mock(OperationContext.class), new HashMap<>(), "securityDomain", false); + remotingConnection, mockServer, RandomUtil.randomUUIDSimpleString(), Mockito.mock(SessionCallback.class), Mockito.mock(OperationContext.class), new HashMap<>(), new HashMap<>(), "securityDomain", false); try { new ServerConsumerImpl(1, session, null, null, 1, true, false, new NullStorageManager(), sessionCallback, true, true, managementService, false, 0, server);