You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/07/24 02:27:23 UTC
[activemq-artemis] branch master updated: ARTEMIS-2848 RA fails
w/durable sub w/legacy prefix
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new 52d3ed1 ARTEMIS-2848 RA fails w/durable sub w/legacy prefix
new 6e5025f This closes #3222
52d3ed1 is described below
commit 52d3ed1c68a52457982f91bd5371a706eefaa253
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Thu Jul 16 15:02:23 2020 -0500
ARTEMIS-2848 RA fails w/durable sub w/legacy prefix
---
.../artemis/ra/ActiveMQResourceAdapter.java | 2 +-
.../artemis/ra/inflow/ActiveMQMessageHandler.java | 3 +-
.../integration/ra/ActiveMQMessageHandlerTest.java | 53 ++++++++++++++++++++++
3 files changed, 56 insertions(+), 2 deletions(-)
diff --git a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/ActiveMQResourceAdapter.java b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/ActiveMQResourceAdapter.java
index ba6127b..4d19cfd 100644
--- a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/ActiveMQResourceAdapter.java
+++ b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/ActiveMQResourceAdapter.java
@@ -1796,7 +1796,7 @@ public class ActiveMQResourceAdapter implements ResourceAdapter, Serializable {
cf.setUseTopologyForLoadBalancing(raProperties.isUseTopologyForLoadBalancing());
cf.setEnableSharedClientID(true);
- cf.setEnable1xPrefixes(raProperties.isEnable1xPrefixes() == null ? false : raProperties.isEnable1xPrefixes());
+ cf.setEnable1xPrefixes(overrideProperties.isEnable1xPrefixes() != null ? overrideProperties.isEnable1xPrefixes() : raProperties.isEnable1xPrefixes() == null ? false : raProperties.isEnable1xPrefixes());
setParams(cf, overrideProperties);
return cf;
}
diff --git a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
index 1cb6bb4..68bfe14 100644
--- a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
+++ b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
@@ -41,6 +41,7 @@ import org.apache.activemq.artemis.api.core.client.MessageHandler;
import org.apache.activemq.artemis.core.client.impl.ClientConsumerInternal;
import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryInternal;
import org.apache.activemq.artemis.core.client.impl.ClientSessionInternal;
+import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.jms.client.ActiveMQMessage;
import org.apache.activemq.artemis.jms.client.ConnectionFactoryOptions;
@@ -140,7 +141,7 @@ public class ActiveMQMessageHandler implements MessageHandler, FailoverEventList
boolean selectorChanged = selector == null && oldFilterString != null || oldFilterString == null && selector != null || (oldFilterString != null && selector != null && !oldFilterString.toString().equals(selector));
- SimpleString oldTopicName = subResponse.getAddress();
+ SimpleString oldTopicName = (enable1XPrefix ? PacketImpl.OLD_TOPIC_PREFIX : SimpleString.toSimpleString("")).concat(subResponse.getAddress());
boolean topicChanged = !oldTopicName.equals(activation.getAddress());
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ra/ActiveMQMessageHandlerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ra/ActiveMQMessageHandlerTest.java
index d372054..776187c 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ra/ActiveMQMessageHandlerTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ra/ActiveMQMessageHandlerTest.java
@@ -86,6 +86,59 @@ public class ActiveMQMessageHandlerTest extends ActiveMQRATestBase {
}
@Test
+ public void testDurableTopicSubscriptionWith1xPrefixesOnSpec() throws Exception {
+ internalTestDurableTopicSubscriptionWith1xPrefixes(false);
+ }
+
+ @Test
+ public void testDurableTopicSubscriptionWith1xPrefixesOnRA() throws Exception {
+ internalTestDurableTopicSubscriptionWith1xPrefixes(true);
+ }
+
+ public void internalTestDurableTopicSubscriptionWith1xPrefixes(boolean ra) throws Exception {
+ server.getRemotingService().createAcceptor("test", "tcp://localhost:61617?anycastPrefix=jms.queue.;multicastPrefix=jms.topic.").start();
+ ActiveMQResourceAdapter qResourceAdapter = newResourceAdapter();
+ if (ra) {
+ qResourceAdapter.setEnable1xPrefixes(true);
+ }
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ ActiveMQActivationSpec spec = new ActiveMQActivationSpec();
+ spec.setSetupAttempts(1);
+ spec.setSetupInterval(500L);
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Topic");
+ spec.setDestination("jms.topic.MyTopic");
+ if (!ra) {
+ spec.setEnable1xPrefixes(true);
+ }
+ spec.setSubscriptionDurability("Durable");
+ spec.setClientId("myClientId");
+ spec.setSubscriptionName("mySubscriptionName");
+ qResourceAdapter.setConnectorClassName(NETTY_CONNECTOR_FACTORY);
+ qResourceAdapter.setConnectionParameters("host=localhost;port=61617");
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = locator.createSessionFactory().createSession();
+ ClientProducer clientProducer = session.createProducer("MyTopic");
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("teststring");
+ clientProducer.send(message);
+ session.close();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
+
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+
+ qResourceAdapter.stop();
+ }
+
+ @Test
public void testObjectMessageReceiveSerializationControl() throws Exception {
String blackList = "org.apache.activemq.artemis.tests.integration.ra";
String whiteList = "*";