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-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..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 @@ -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 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 setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.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-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/OpenWireProtocolManager.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java index 94de8c5ab87..d39daf9e731 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 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 setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.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-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 5bd009477df..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); 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..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 @@ -245,7 +245,7 @@ private StompSession internalGetSession(StompConnection connection, Map 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 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 setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.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/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/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..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 @@ -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 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 setTemporaryMulticastPrefix(String temporaryMulticastPrefix) { + for (String prefix : temporaryMulticastPrefix.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..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,8 +75,14 @@ default void removeHandler(String name) { void setMulticastPrefix(String multicastPrefix); + void setTemporaryAnycastPrefix(String temporaryAnycastPrefix); + + void setTemporaryMulticastPrefix(String temporaryMulticastPrefix); + Map getPrefixes(); + Map getTemporaryPrefixes(); + void setSecurityDomain(String securityDomain); String getSecurityDomain(); 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); 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..466d67ccdaf --- /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 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 + * 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 + + "&temporaryAnycastPrefix=" + 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 temporaryAnycastPrefix should create an ANYCAST queue"); + assertTrue(queue.isTemporary(), "Queue created by temporaryAnycastPrefix 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 + + "&temporaryAnycastPrefix=" + 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 + + "&temporaryAnycastPrefix=" + 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 + + "&temporaryMulticastPrefix=" + 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 temporaryMulticastPrefix should create an address"); + assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST), + "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)); + 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 + + "&temporaryMulticastPrefix=" + 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 + + "&temporaryMulticastPrefix=" + 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 + + "&temporaryAnycastPrefix=" + TEMP_QUEUE_PREFIX + + "&temporaryMulticastPrefix=" + 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 + + "&temporaryAnycastPrefix=" + 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 temporaryAnycastPrefix"); + + 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(); + } +}