Skip to content

Commit 5317337

Browse files
jbertramtabish121
jbertram
authored and
tabish121
committed
ARTEMIS-5996 refactor STOMP to use checkAutoCreate from ServerSession
The STOMP implementation duplicates much of the logic related to auto-creating addresses & queues from the ServerSession. It should be refactored to use ServerSession's checkAutoCreate method.
1 parent 845339b commit 5317337

7 files changed

Lines changed: 129 additions & 91 deletions

File tree

‎artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/ActiveMQStompProtocolMessageBundle.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
/**
2424
* Logger Codes 339000 - 339999
2525
*/
26-
@LogBundle(projectCode = "AMQ", regexID = "339[0-9]{3}")
26+
@LogBundle(projectCode = "AMQ", regexID = "339[0-9]{3}", retiredIDs = {339041})
2727
public interface ActiveMQStompProtocolMessageBundle {
2828

2929
ActiveMQStompProtocolMessageBundle BUNDLE = BundleFactory.newBundle(ActiveMQStompProtocolMessageBundle.class);
@@ -144,7 +144,4 @@ public interface ActiveMQStompProtocolMessageBundle {
144144

145145
@Message(id = 339040, value = "Undefined escape sequence: {}")
146146
ActiveMQStompException undefinedEscapeSequence(String sequence);
147-
148-
@Message(id = 339041, value = "Not allowed to specify {} semantics on {} address.")
149-
ActiveMQStompException illegalSemantics(String requested, String exists);
150147
}

‎artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java‎

Lines changed: 47 additions & 79 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
import java.lang.invoke.MethodHandles;
2020
import java.util.ArrayList;
2121
import java.util.Collections;
22-
import java.util.EnumSet;
2322
import java.util.HashSet;
2423
import java.util.List;
2524
import java.util.Objects;
@@ -33,12 +32,11 @@
3332
import org.apache.activemq.artemis.api.core.ActiveMQAddressDoesNotExistException;
3433
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
3534
import org.apache.activemq.artemis.api.core.ActiveMQException;
36-
import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
3735
import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
36+
import org.apache.activemq.artemis.api.core.AutoCreateResult;
3837
import org.apache.activemq.artemis.api.core.ICoreMessage;
3938
import org.apache.activemq.artemis.api.core.QueueConfiguration;
4039
import org.apache.activemq.artemis.api.core.RoutingType;
41-
import org.apache.activemq.artemis.api.core.SimpleString;
4240
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
4341
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
4442
import org.apache.activemq.artemis.core.protocol.stomp.v10.StompFrameHandlerV10;
@@ -47,16 +45,12 @@
4745
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
4846
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
4947
import org.apache.activemq.artemis.core.server.ServerConsumer;
50-
import org.apache.activemq.artemis.core.server.ServerSession;
51-
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
52-
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
5348
import org.apache.activemq.artemis.selector.filter.FilterException;
5449
import org.apache.activemq.artemis.selector.impl.SelectorParser;
5550
import org.apache.activemq.artemis.spi.core.protocol.AbstractRemotingConnection;
5651
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
5752
import org.apache.activemq.artemis.spi.core.remoting.Connection;
5853
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
59-
import org.apache.activemq.artemis.utils.CompositeAddress;
6054
import org.apache.activemq.artemis.utils.ConfigurationHelper;
6155
import org.apache.activemq.artemis.utils.ExecutorFactory;
6256
import org.apache.activemq.artemis.utils.VersionLoader;
@@ -166,63 +160,6 @@ public boolean hasBytes() {
166160
this.minLargeMessageSize = ConfigurationHelper.getIntProperty(TransportConstants.STOMP_MIN_LARGE_MESSAGE_SIZE, ConfigurationHelper.getIntProperty(TransportConstants.STOMP_MIN_LARGE_MESSAGE_SIZE_DEPRECATED, ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, acceptorUsed.getConfiguration()), acceptorUsed.getConfiguration());
167161
}
168162

169-
// TODO this should take a type - send or receive so it knows whether to check the address or the queue
170-
public void checkDestination(String destination) throws ActiveMQStompException {
171-
if (!manager.destinationExists(destination)) {
172-
throw BUNDLE.destinationNotExist(destination).setHandler(frameHandler);
173-
}
174-
}
175-
176-
public void autoCreateDestinationIfPossible(String destination, RoutingType routingType) throws ActiveMQStompException {
177-
try {
178-
SimpleString simpleDestination = SimpleString.of(destination);
179-
AddressInfo addressInfo = manager.getServer().getAddressInfo(simpleDestination);
180-
AddressSettings addressSettings = manager.getServer().getAddressSettingsRepository().getMatch(destination);
181-
RoutingType effectiveAddressRoutingType = Objects.requireNonNullElse(routingType, addressSettings.getDefaultAddressRoutingType());
182-
ServerSession session = getSession().getCoreSession();
183-
/*
184-
* If the address doesn't exist then it is created if possible.
185-
* If the address does exist but doesn't support the routing-type then the address is updated if possible.
186-
*/
187-
if (addressInfo == null) {
188-
if (addressSettings.isAutoCreateAddresses()) {
189-
session.createAddress(simpleDestination, effectiveAddressRoutingType, true);
190-
}
191-
} else if (!addressInfo.getRoutingTypes().contains(effectiveAddressRoutingType)) {
192-
if (addressSettings.isAutoCreateAddresses()) {
193-
EnumSet<RoutingType> routingTypes = EnumSet.noneOf(RoutingType.class);
194-
for (RoutingType existingRoutingType : addressInfo.getRoutingTypes()) {
195-
routingTypes.add(existingRoutingType);
196-
}
197-
routingTypes.add(effectiveAddressRoutingType);
198-
manager.getServer().updateAddressInfo(simpleDestination, routingTypes);
199-
}
200-
}
201-
202-
// auto create the queue if the address is ANYCAST or FQQN
203-
if ((CompositeAddress.isFullyQualified(destination) || effectiveAddressRoutingType == RoutingType.ANYCAST) && addressSettings.isAutoCreateQueues() && manager.getServer().locateQueue(simpleDestination) == null) {
204-
session.createQueue(QueueConfiguration.of(destination).setRoutingType(effectiveAddressRoutingType).setAutoCreated(true));
205-
}
206-
} catch (ActiveMQQueueExistsException e) {
207-
// ignore
208-
} catch (Exception e) {
209-
logger.debug("Exception while auto-creating destination", e);
210-
throw new ActiveMQStompException(e.getMessage(), e).setHandler(frameHandler);
211-
}
212-
}
213-
214-
public void checkRoutingSemantics(String destination, RoutingType routingType) throws ActiveMQStompException {
215-
AddressInfo addressInfo = manager.getServer().getAddressInfo(SimpleString.of(destination));
216-
217-
// may be null here if, for example, the management address is being checked
218-
if (addressInfo != null) {
219-
Set<RoutingType> actualDeliveryModesOfAddress = addressInfo.getRoutingTypes();
220-
if (routingType != null && !actualDeliveryModesOfAddress.contains(routingType)) {
221-
throw BUNDLE.illegalSemantics(routingType.toString(), actualDeliveryModesOfAddress.toString());
222-
}
223-
}
224-
}
225-
226163
@Override
227164
public void destroy() {
228165
if (DESTROYED_UPDATER.compareAndSet(this, 0, 1)) {
@@ -551,9 +488,48 @@ StompPostReceiptFunction subscribe(String destination,
551488
RoutingType subscriptionType,
552489
Integer consumerWindowSize) throws ActiveMQStompException {
553490
validateSelector(selector);
554-
autoCreateDestinationIfPossible(destination, subscriptionType);
555-
checkDestination(destination);
556-
checkRoutingSemantics(destination, subscriptionType);
491+
checkAutoCreate(destination, subscriptionType);
492+
String subscriptionID = getSubscriptionID(destination, id);
493+
494+
try {
495+
return manager.subscribe(this,
496+
subscriptionID,
497+
durableSubscriptionName,
498+
destination,
499+
getSelector(selector, noLocal),
500+
Objects.requireNonNullElse(ack, Stomp.Headers.Subscribe.AckModeValues.AUTO),
501+
noLocal,
502+
consumerWindowSize);
503+
} catch (ActiveMQStompException e) {
504+
throw e;
505+
} catch (Exception e) {
506+
throw BUNDLE.errorCreatingSubscription(subscriptionID, e).setHandler(frameHandler);
507+
}
508+
}
509+
510+
protected void checkAutoCreate(String destination, RoutingType subscriptionType) throws ActiveMQStompException {
511+
AutoCreateResult autoCreateResult;
512+
try {
513+
RoutingType routingType = getSubscriptionRoutingType(destination, subscriptionType);
514+
autoCreateResult = getSession().getCoreSession().checkAutoCreate(QueueConfiguration.of(destination).setRoutingType(routingType));
515+
} catch (Exception e) {
516+
logger.debug("Exception while auto-creating destination", e);
517+
throw new ActiveMQStompException(e.getMessage(), e).setHandler(frameHandler);
518+
}
519+
if (autoCreateResult == AutoCreateResult.NOT_FOUND) {
520+
throw BUNDLE.destinationNotExist(destination).setHandler(frameHandler);
521+
}
522+
}
523+
524+
private RoutingType getSubscriptionRoutingType(String destination, RoutingType subscriptionType) {
525+
if (subscriptionType == null) {
526+
return getManager().getServer().getAddressSettingsRepository().getMatch(destination).getDefaultAddressRoutingType();
527+
} else {
528+
return subscriptionType;
529+
}
530+
}
531+
532+
private String getSelector(String selector, boolean noLocal) {
557533
if (noLocal) {
558534
String noLocalFilter = "(" + CONNECTION_ID_PROPERTY_NAME_STRING + " <> '" + getID().toString() + "' OR " + CONNECTION_ID_PROPERTY_NAME_STRING + " IS NULL)";
559535
if (selector == null) {
@@ -562,11 +538,10 @@ StompPostReceiptFunction subscribe(String destination,
562538
selector = "(" + selector + ") AND " + noLocalFilter;
563539
}
564540
}
541+
return selector;
542+
}
565543

566-
if (ack == null) {
567-
ack = Stomp.Headers.Subscribe.AckModeValues.AUTO;
568-
}
569-
544+
private String getSubscriptionID(String destination, String id) throws ActiveMQStompException {
570545
String subscriptionID = null;
571546
if (id != null) {
572547
subscriptionID = id;
@@ -576,14 +551,7 @@ StompPostReceiptFunction subscribe(String destination,
576551
}
577552
subscriptionID = "subscription/" + destination;
578553
}
579-
580-
try {
581-
return manager.subscribe(this, subscriptionID, durableSubscriptionName, destination, selector, ack, noLocal, consumerWindowSize);
582-
} catch (ActiveMQStompException e) {
583-
throw e;
584-
} catch (Exception e) {
585-
throw BUNDLE.errorCreatingSubscription(subscriptionID, e).setHandler(frameHandler);
586-
}
554+
return subscriptionID;
587555
}
588556

589557
private void validateSelector(String selector) throws ActiveMQStompException {

‎artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -193,9 +193,7 @@ public StompFrame onSend(StompFrame frame) {
193193
connection.validate();
194194
String destination = getDestination(frame);
195195
RoutingType routingType = getRoutingType(frame.getHeader(Headers.Send.DESTINATION_TYPE), frame.getHeader(Headers.Send.DESTINATION));
196-
connection.autoCreateDestinationIfPossible(destination, routingType);
197-
connection.checkDestination(destination);
198-
connection.checkRoutingSemantics(destination, routingType);
196+
connection.checkAutoCreate(destination, routingType);
199197
String txID = frame.getHeader(Stomp.Headers.TRANSACTION);
200198

201199
long timestamp = System.currentTimeMillis();

‎tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompLVQTest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323

2424
import org.apache.activemq.artemis.api.core.Message;
2525
import org.apache.activemq.artemis.api.core.QueueConfiguration;
26+
import org.apache.activemq.artemis.api.core.RoutingType;
2627
import org.apache.activemq.artemis.core.protocol.stomp.Stomp;
2728
import org.apache.activemq.artemis.tests.integration.stomp.util.ClientStompFrame;
2829
import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnection;
@@ -53,7 +54,7 @@ public StompLVQTest() {
5354
public void setUp() throws Exception {
5455
super.setUp();
5556

56-
server.createQueue(QueueConfiguration.of(queue).setLastValue(true).setExclusive(true));
57+
server.createQueue(QueueConfiguration.of(queue).setLastValue(true).setExclusive(true).setRoutingType(RoutingType.ANYCAST));
5758

5859
producerConn = StompClientConnectionFactory.createClientConnection(uri);
5960
consumerConn = StompClientConnectionFactory.createClientConnection(uri);

‎tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestBase.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,10 @@ public abstract class StompTestBase extends ActiveMQTestBase {
9393

9494
protected String defPass = "wombats";
9595

96+
protected String onlySendCredential = "onlySend";
97+
98+
protected String onlyConsumeCredential = "onlyConsume";
99+
96100
public StompTestBase(String scheme) {
97101
this.scheme = scheme;
98102
}
@@ -211,6 +215,16 @@ protected ActiveMQServer createServer() throws Exception {
211215
final String role = "testRole";
212216
securityManager.getConfiguration().addRole(defUser, role);
213217
config.getSecurityRoles().put("#", new HashSet<Role>(Set.of(new Role(role, true, true, true, true, true, true, true, true, true, true, false, false))));
218+
219+
final String onlySend = onlySendCredential;
220+
securityManager.getConfiguration().addUser(onlySend, onlySend);
221+
securityManager.getConfiguration().addRole(onlySend, onlySend);
222+
config.getSecurityRoles().get("#").add(new Role(onlySend, true, false, false, false, false, false, false, false, false, false, false, false));
223+
224+
final String onlyConsume = onlyConsumeCredential;
225+
securityManager.getConfiguration().addUser(onlyConsume, onlyConsume);
226+
securityManager.getConfiguration().addRole(onlyConsume, onlyConsume);
227+
config.getSecurityRoles().get("#").add(new Role(onlyConsume, false, true, false, false, false, false, false, false, false, false, false, false));
214228
}
215229

216230
return activeMQServer;

‎tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompWithSecurityTest.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,18 @@
1818

1919
import javax.jms.MessageConsumer;
2020
import javax.jms.TextMessage;
21+
import java.util.UUID;
2122

23+
import org.apache.activemq.artemis.api.core.RoutingType;
24+
import org.apache.activemq.artemis.api.core.SimpleString;
25+
import org.apache.activemq.artemis.core.protocol.stomp.Stomp;
2226
import org.apache.activemq.artemis.tests.integration.stomp.util.ClientStompFrame;
2327
import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnection;
2428
import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnectionFactory;
2529
import org.junit.jupiter.api.Test;
2630

2731
import static org.junit.jupiter.api.Assertions.assertEquals;
32+
import static org.junit.jupiter.api.Assertions.assertFalse;
2833
import static org.junit.jupiter.api.Assertions.assertNotNull;
2934
import static org.junit.jupiter.api.Assertions.assertTrue;
3035

@@ -68,4 +73,57 @@ public void testJMSXUserID() throws Exception {
6873
long tmsg = message.getJMSTimestamp();
6974
assertTrue(Math.abs(tnow - tmsg) < 1000);
7075
}
76+
77+
@Test
78+
public void testSendMessageWithDifferentRoutingType() throws Exception {
79+
// validate presuppositions
80+
SimpleString queueName = SimpleString.of(getQueuePrefix() + getQueueName());
81+
assertNotNull(server.locateQueue(queueName));
82+
assertTrue(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.ANYCAST));
83+
assertFalse(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.MULTICAST));
84+
85+
StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
86+
conn.connect(onlySendCredential, onlySendCredential);
87+
88+
ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND);
89+
frame.addHeader(Stomp.Headers.Send.DESTINATION, queueName.toString());
90+
frame.setBody("Hello World");
91+
frame.addHeader(Stomp.Headers.Send.DESTINATION_TYPE, RoutingType.MULTICAST.toString());
92+
frame.addHeader(Stomp.Headers.RECEIPT_REQUESTED, UUID.randomUUID().toString());
93+
ClientStompFrame result = conn.sendFrame(frame);
94+
assertNotNull(result);
95+
assertEquals(Stomp.Responses.ERROR, result.getCommand());
96+
assertTrue(result.getHeader(Stomp.Headers.Error.MESSAGE).contains("AMQ229032"));
97+
98+
conn.disconnect();
99+
100+
assertTrue(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.ANYCAST));
101+
assertFalse(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.MULTICAST));
102+
}
103+
104+
@Test
105+
public void testSubscribeWithDifferentRoutingType() throws Exception {
106+
// validate presuppositions
107+
SimpleString queueName = SimpleString.of(getQueuePrefix() + getQueueName());
108+
assertNotNull(server.locateQueue(queueName));
109+
assertTrue(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.ANYCAST));
110+
assertFalse(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.MULTICAST));
111+
112+
StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
113+
conn.connect(onlyConsumeCredential, onlyConsumeCredential);
114+
115+
ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE);
116+
frame.addHeader(Stomp.Headers.Subscribe.DESTINATION, queueName.toString());
117+
frame.addHeader(Stomp.Headers.Subscribe.SUBSCRIPTION_TYPE, RoutingType.MULTICAST.toString());
118+
frame.addHeader(Stomp.Headers.RECEIPT_REQUESTED, UUID.randomUUID().toString());
119+
ClientStompFrame result = conn.sendFrame(frame);
120+
assertNotNull(result);
121+
assertEquals(Stomp.Responses.ERROR, result.getCommand());
122+
assertTrue(result.getHeader(Stomp.Headers.Error.MESSAGE).contains("AMQ229032"));
123+
124+
conn.disconnect();
125+
126+
assertTrue(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.ANYCAST));
127+
assertFalse(server.getAddressInfo(queueName).getRoutingTypes().contains(RoutingType.MULTICAST));
128+
}
71129
}

‎tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v12/StompV12Test.java‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -399,8 +399,9 @@ public void testHeaderRepetitive() throws Exception {
399399
String cLen = String.valueOf(body.getBytes(StandardCharsets.UTF_8).length);
400400

401401
ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND)
402-
.addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName())
403-
.addHeader(Stomp.Headers.Subscribe.DESTINATION, "aNonexistentQueue")
402+
.addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName())
403+
.addHeader(Stomp.Headers.Send.DESTINATION, "aNonexistentQueue")
404+
.addHeader(Stomp.Headers.Send.DESTINATION_TYPE, RoutingType.ANYCAST.toString())
404405
.addHeader(Stomp.Headers.CONTENT_TYPE, "application/xml")
405406
.addHeader(Stomp.Headers.CONTENT_LENGTH, cLen)
406407
.addHeader("foo", "value1")
@@ -435,8 +436,9 @@ public void testHeaderRepetitive() throws Exception {
435436
cLen = String.valueOf(body.getBytes(StandardCharsets.UTF_8).length);
436437

437438
frame = conn.createFrame(Stomp.Commands.SEND)
438-
.addHeader(Stomp.Headers.Subscribe.DESTINATION, "aNonexistentQueue")
439-
.addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName())
439+
.addHeader(Stomp.Headers.Send.DESTINATION, "aNonexistentQueue")
440+
.addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName())
441+
.addHeader(Stomp.Headers.Send.DESTINATION_TYPE, RoutingType.ANYCAST.toString())
440442
.addHeader(Stomp.Headers.CONTENT_TYPE, "application/xml")
441443
.addHeader(Stomp.Headers.CONTENT_LENGTH, cLen)
442444
.addHeader(Stomp.Headers.RECEIPT_REQUESTED, "1234")

0 commit comments

Comments
 (0)