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 2018/10/10 22:48:17 UTC
[1/2] activemq-artemis git commit: ARTEMIS-2103 - use the full
openwire consumer queue for the mapped virtual topic queue binding,
fix and test
Repository: activemq-artemis
Updated Branches:
refs/heads/master 714a3f862 -> 76595e79a
ARTEMIS-2103 - use the full openwire consumer queue for the mapped virtual topic queue binding, fix and test
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/b812bfdb
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/b812bfdb
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/b812bfdb
Branch: refs/heads/master
Commit: b812bfdbedb0b5fcf8492c652ca5badb86b8ce9e
Parents: 714a3f8
Author: gtully <ga...@gmail.com>
Authored: Wed Oct 3 13:24:42 2018 +0100
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Oct 10 18:48:11 2018 -0400
----------------------------------------------------------------------
.../openwire/OpenWireProtocolManager.java | 4 +-
docs/migration-guide/en/VirtualTopics.md | 4 +-
docs/user-manual/en/openwire.md | 4 +-
.../openwire/virtual-topic-mapping/readme.md | 2 +-
.../jms/example/VirtualTopicMappingExample.java | 2 +-
.../VirtualTopicToFQQNOpenWireTest.java | 138 +++++++++++++++++++
6 files changed, 146 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b812bfdb/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java
----------------------------------------------------------------------
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 95a400e..505564d 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
@@ -657,8 +657,8 @@ public class OpenWireProtocolManager implements ProtocolManager<Interceptor>, Cl
fqqn.append(paths[i]);
}
fqqn.append(CompositeAddress.SEPARATOR);
- // consumer queue
- for (int i = 0; i < filterPathTerminus; i++) {
+ // consumer queue - the full vt queue
+ for (int i = 0; i < paths.length; i++) {
if (i > 0) {
fqqn.append(ActiveMQDestination.PATH_SEPERATOR);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b812bfdb/docs/migration-guide/en/VirtualTopics.md
----------------------------------------------------------------------
diff --git a/docs/migration-guide/en/VirtualTopics.md b/docs/migration-guide/en/VirtualTopics.md
index 6ad183d..b98ac80 100644
--- a/docs/migration-guide/en/VirtualTopics.md
+++ b/docs/migration-guide/en/VirtualTopics.md
@@ -33,7 +33,7 @@ For example, a default 5.x consumer destination for topic `VirtualTopic.Orders`
would be replaced with an Artemis FQQN comprised of the address and queue.
```
...
- Queue subscriptionQueue = session.createQueue("VirtualTopic.Orders::Consumer.A");
+ Queue subscriptionQueue = session.createQueue("VirtualTopic.Orders::Consumer.A.VirtualTopic.Orders");
session.createConsumer(subscriptionQueue);
```
@@ -49,7 +49,7 @@ E.g: For the default 5.x virtual topic consumer prefix of ```Consumer.*.``` the
However, there is a caveat because this value needs to be encoded in a uri for the xml configuration. Any unsafe url characters
, in this case: ```> ;``` need to be escaped with their hex code point representation; leading to a value of ```Consumer.*.%3E%3B2```.
In this way a consumer destination of ```Consumer.A.VirtualTopic.Orders``` will be transformed into a FQQN of
-```VirtualTopic.Orders::Consumer.A```.
+```VirtualTopic.Orders::Consumer.A.VirtualTopic.Orders```.
Durable topic subscribers in a network of brokers
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b812bfdb/docs/user-manual/en/openwire.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/openwire.md b/docs/user-manual/en/openwire.md
index 7d24fb8..9cecfe1 100644
--- a/docs/user-manual/en/openwire.md
+++ b/docs/user-manual/en/openwire.md
@@ -85,7 +85,7 @@ The two parameters are configured on an OpenWire `acceptor`, e.g.:
For existing OpenWire consumers of virtual topic destinations it is possible to
configure a mapping function that will translate the virtual topic consumer
-destination into a FQQN address. This address then represents the consumer as a
+destination into a FQQN address. This address will then represents the consumer as a
multicast binding to an address representing the virtual topic.
The configuration string property `virtualTopicConsumerWildcards` has two parts
@@ -103,7 +103,7 @@ this transforms to `Consumer.*.%3E%3B2` when the url significant characters
```
This will translate `Consumer.A.VirtualTopic.Orders` into a FQQN of
-`VirtualTopic.Orders::Consumer.A` using the int component `2` of the
+`VirtualTopic.Orders::Consumer.A.VirtualTopic.Orders` using the int component `2` of the
configuration to identify the consumer queue as the first two paths of the
destination. `virtualTopicConsumerWildcards` is multi valued using a `,`
separator.
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b812bfdb/examples/protocols/openwire/virtual-topic-mapping/readme.md
----------------------------------------------------------------------
diff --git a/examples/protocols/openwire/virtual-topic-mapping/readme.md b/examples/protocols/openwire/virtual-topic-mapping/readme.md
index e96724d..4660492 100644
--- a/examples/protocols/openwire/virtual-topic-mapping/readme.md
+++ b/examples/protocols/openwire/virtual-topic-mapping/readme.md
@@ -18,7 +18,7 @@ Address.
The example sends a message to a topic (using openwire protocol) and an openwire consumer listens on the backing queue
using the ActiveMQ 5.x virtual topic naming convention. Due to the acceptor url parameter `virtualTopicConsumerWildcards`,
(see below), Artemis maps the consumer consuming from `Consumer.A.VirtualTopic.Orders` to actually consume from
-FQQN of `VirtualTopic.Orders::Consumer.A`
+FQQN of `VirtualTopic.Orders::Consumer.A.VirtualTopic.Orders`
```xml
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b812bfdb/examples/protocols/openwire/virtual-topic-mapping/src/main/java/org/apache/activemq/artemis/jms/example/VirtualTopicMappingExample.java
----------------------------------------------------------------------
diff --git a/examples/protocols/openwire/virtual-topic-mapping/src/main/java/org/apache/activemq/artemis/jms/example/VirtualTopicMappingExample.java b/examples/protocols/openwire/virtual-topic-mapping/src/main/java/org/apache/activemq/artemis/jms/example/VirtualTopicMappingExample.java
index eff7b18..fd849b5 100644
--- a/examples/protocols/openwire/virtual-topic-mapping/src/main/java/org/apache/activemq/artemis/jms/example/VirtualTopicMappingExample.java
+++ b/examples/protocols/openwire/virtual-topic-mapping/src/main/java/org/apache/activemq/artemis/jms/example/VirtualTopicMappingExample.java
@@ -31,7 +31,7 @@ import org.apache.activemq.ActiveMQConnectionFactory;
* The example sends a message to a topic (using openwire protocol) and an openwire consumer listens on the backing queue
* using the ActiveMQ 5.x virtual topic naming convention. Due to the acceptor parameter virtualTopicConsumerWildcards
* Artemis maps the consumer consuming from "Consumer.A.VirtualTopic.Orders" to actually consume from
- * FQQN "VirtualTopic.Orders::Consumer.A"
+ * FQQN "VirtualTopic.Orders::Consumer.A.VirtualTopic.Orders"
*/
public class VirtualTopicMappingExample {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b812bfdb/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/VirtualTopicToFQQNOpenWireTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/VirtualTopicToFQQNOpenWireTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/VirtualTopicToFQQNOpenWireTest.java
index 34d08bc..228c904 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/VirtualTopicToFQQNOpenWireTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/VirtualTopicToFQQNOpenWireTest.java
@@ -84,4 +84,142 @@ public class VirtualTopicToFQQNOpenWireTest extends OpenWireTestBase {
}
}
}
+
+ @Test
+ public void testTwoTopicSubsSameNameAutoVirtualTopicFQQN() throws Exception {
+ Connection connection = null;
+
+ SimpleString topic1 = new SimpleString("VirtualTopic.Orders1");
+ SimpleString topic2 = new SimpleString("VirtualTopic.Orders2");
+
+ this.server.getAddressSettingsRepository().getMatch("VirtualTopic.#").setAutoCreateQueues(true);
+ this.server.getAddressSettingsRepository().getMatch("VirtualTopic.#").setAutoCreateAddresses(true);
+
+ try {
+ ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(urlString);
+ activeMQConnectionFactory.setWatchTopicAdvisories(false);
+ connection = activeMQConnectionFactory.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination1 = session.createTopic(topic1.toString());
+ Destination destination2 = session.createTopic(topic2.toString());
+
+ MessageConsumer messageConsumer1 = session.createConsumer(session.createQueue("Consumer.A." + topic1.toString()));
+ MessageConsumer messageConsumer2 = session.createConsumer(session.createQueue("Consumer.A." + topic2.toString()));
+
+ MessageProducer producer = session.createProducer(null);
+ TextMessage message = session.createTextMessage("This is a text message to 1");
+ producer.send(destination1, message);
+ message = session.createTextMessage("This is a text message to 2");
+ producer.send(destination2, message);
+
+
+ TextMessage messageReceived1 = (TextMessage) messageConsumer1.receive(2000);
+ TextMessage messageReceived2 = (TextMessage) messageConsumer2.receive(2000);
+
+ assertNotNull(messageReceived1);
+ assertNotNull(messageReceived2);
+
+ String text = messageReceived1.getText();
+ assertEquals("This is a text message to 1", text);
+
+ text = messageReceived2.getText();
+ assertEquals("This is a text message to 2", text);
+
+ messageConsumer1.close();
+ messageConsumer2.close();
+
+ } finally {
+ if (connection != null) {
+ connection.close();
+ }
+ }
+ }
+
+
+ @Test
+ public void testAutoVirtualTopicWildcardFQQN() throws Exception {
+ Connection connection = null;
+
+ SimpleString topicA = new SimpleString("VirtualTopic.Orders.A");
+ SimpleString topicB = new SimpleString("VirtualTopic.Orders.B");
+ SimpleString topic = new SimpleString("VirtualTopic.Orders.>");
+
+ this.server.getAddressSettingsRepository().getMatch("VirtualTopic.#").setAutoCreateQueues(true);
+ this.server.getAddressSettingsRepository().getMatch("VirtualTopic.#").setAutoCreateAddresses(true);
+
+ try {
+ ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(urlString);
+ activeMQConnectionFactory.setWatchTopicAdvisories(false);
+ connection = activeMQConnectionFactory.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createTopic(topicA.toString() + "," + topicB.toString());
+
+ MessageConsumer messageConsumerA = session.createConsumer(session.createQueue("Consumer.A." + topic.toString()));
+ // MessageConsumer messageConsumerB = session.createConsumer(session.createQueue("Consumer.B." + topic.toString()));
+
+ MessageProducer producer = session.createProducer(destination);
+ TextMessage message = session.createTextMessage("This is a text message");
+ producer.send(message);
+
+ TextMessage messageReceivedA = (TextMessage) messageConsumerA.receive(2000);
+ TextMessage messageReceivedB = (TextMessage) messageConsumerA.receive(2000);
+
+ assertTrue((messageReceivedA != null && messageReceivedB != null));
+ String text = messageReceivedA.getText();
+ assertEquals("This is a text message", text);
+
+ messageConsumerA.close();
+
+ } finally {
+ if (connection != null) {
+ connection.close();
+ }
+ }
+ }
+
+ @Test
+ public void testAutoVirtualTopicWildcardStarFQQN() throws Exception {
+ Connection connection = null;
+
+ SimpleString topicA = new SimpleString("VirtualTopic.Orders.A");
+ SimpleString topicB = new SimpleString("VirtualTopic.Orders.B");
+ SimpleString topic = new SimpleString("VirtualTopic.Orders.*");
+
+ this.server.getAddressSettingsRepository().getMatch("VirtualTopic.#").setAutoCreateQueues(true);
+ this.server.getAddressSettingsRepository().getMatch("VirtualTopic.#").setAutoCreateAddresses(true);
+
+ try {
+ ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(urlString);
+ activeMQConnectionFactory.setWatchTopicAdvisories(false);
+ connection = activeMQConnectionFactory.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createTopic(topicA.toString() + "," + topicB.toString());
+
+ MessageConsumer messageConsumerA = session.createConsumer(session.createQueue("Consumer.A." + topic.toString()));
+
+ MessageProducer producer = session.createProducer(destination);
+ TextMessage message = session.createTextMessage("This is a text message");
+ producer.send(message);
+
+ TextMessage messageReceivedA = (TextMessage) messageConsumerA.receive(2000);
+ TextMessage messageReceivedB = (TextMessage) messageConsumerA.receive(2000);
+
+ assertTrue((messageReceivedA != null && messageReceivedB != null));
+ String text = messageReceivedA.getText();
+ assertEquals("This is a text message", text);
+
+ messageConsumerA.close();
+
+ } finally {
+ if (connection != null) {
+ connection.close();
+ }
+ }
+ }
}
[2/2] activemq-artemis git commit: This closes #2341
Posted by cl...@apache.org.
This closes #2341
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/76595e79
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/76595e79
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/76595e79
Branch: refs/heads/master
Commit: 76595e79a1582b1d4e8aed7a95b8e0bd4cd62797
Parents: 714a3f8 b812bfd
Author: Clebert Suconic <cl...@apache.org>
Authored: Wed Oct 10 18:48:12 2018 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Oct 10 18:48:12 2018 -0400
----------------------------------------------------------------------
.../openwire/OpenWireProtocolManager.java | 4 +-
docs/migration-guide/en/VirtualTopics.md | 4 +-
docs/user-manual/en/openwire.md | 4 +-
.../openwire/virtual-topic-mapping/readme.md | 2 +-
.../jms/example/VirtualTopicMappingExample.java | 2 +-
.../VirtualTopicToFQQNOpenWireTest.java | 138 +++++++++++++++++++
6 files changed, 146 insertions(+), 8 deletions(-)
----------------------------------------------------------------------