[
https://issues.apache.org/jira/browse/CAMEL-12624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16543331#comment-16543331
]
ASF GitHub Bot commented on CAMEL-12624:
----------------------------------------
dmvolod closed pull request #2425: CAMEL-12624: ActiveMQ Artemis AMQP
integration issue with topic prefix hardcoded
URL: https://github.com/apache/camel/pull/2425
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/components/camel-amqp/pom.xml b/components/camel-amqp/pom.xml
index 28e85b0e000..92b3bf2fd15 100644
--- a/components/camel-amqp/pom.xml
+++ b/components/camel-amqp/pom.xml
@@ -96,6 +96,18 @@
<artifactId>activemq-broker</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>artemis-server</artifactId>
+ <version>${activemq-artemis-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>artemis-amqp-protocol</artifactId>
+ <version>${activemq-artemis-version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-amqp</artifactId>
diff --git
a/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPComponent.java
b/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPComponent.java
index 43aebab1d86..cc1bda660bb 100644
---
a/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPComponent.java
+++
b/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPComponent.java
@@ -56,7 +56,9 @@ protected void doStart() throws Exception {
if (connectionDetails.size() == 1) {
AMQPConnectionDetails details =
connectionDetails.iterator().next();
JmsConnectionFactory connectionFactory = new
JmsConnectionFactory(details.username(), details.password(), details.uri());
- connectionFactory.setTopicPrefix("topic://");
+ if (details.setTopicPrefix()) {
+ connectionFactory.setTopicPrefix("topic://");
+ }
setConnectionFactory(connectionFactory);
}
super.doStart();
diff --git
a/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPConnectionDetails.java
b/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPConnectionDetails.java
index 029e17b781b..66732e6c1f9 100644
---
a/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPConnectionDetails.java
+++
b/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPConnectionDetails.java
@@ -28,17 +28,29 @@
public static final String AMQP_USERNAME = "AMQP_SERVICE_USERNAME";
public static final String AMQP_PASSWORD = "AMQP_SERVICE_PASSWORD";
+
+ public static final String AMQP_SET_TOPIC_PREFIX = "AMQP_SET_TOPIC_PREFIX";
private final String uri;
private final String username;
private final String password;
+
+ private final boolean setTopicPrefix;
public AMQPConnectionDetails(String uri, String username, String password)
{
this.uri = uri;
this.username = username;
this.password = password;
+ this.setTopicPrefix = true;
+ }
+
+ public AMQPConnectionDetails(String uri, String username, String password,
boolean setTopicPrefix) {
+ this.uri = uri;
+ this.username = username;
+ this.password = password;
+ this.setTopicPrefix = setTopicPrefix;
}
public AMQPConnectionDetails(String uri) {
@@ -53,8 +65,9 @@ public static AMQPConnectionDetails discoverAMQP(CamelContext
camelContext) {
int port = Integer.parseInt(property(propertiesComponent,
AMQP_PORT, "5672"));
String username = property(propertiesComponent, AMQP_USERNAME,
null);
String password = property(propertiesComponent, AMQP_PASSWORD,
null);
+ boolean setTopicPrefix =
Boolean.parseBoolean(property(propertiesComponent, AMQP_SET_TOPIC_PREFIX,
"true"));
- return new AMQPConnectionDetails("amqp://" + host + ":" + port,
username, password);
+ return new AMQPConnectionDetails("amqp://" + host + ":" + port,
username, password, setTopicPrefix);
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -71,6 +84,10 @@ public String username() {
public String password() {
return password;
}
+
+ public boolean setTopicPrefix() {
+ return setTopicPrefix;
+ }
// Helpers
diff --git
a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java
b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java
new file mode 100644
index 00000000000..563aa466735
--- /dev/null
+++
b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java
@@ -0,0 +1,117 @@
+/**
+ * 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.camel.component.amqp.artemis;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
+import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
+import org.apache.activemq.artemis.core.server.embedded.EmbeddedActiveMQ;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.amqp.AMQPComponent;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.impl.PropertyPlaceholderDelegateRegistry;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.apache.camel.component.amqp.AMQPConnectionDetails.AMQP_PORT;
+import static
org.apache.camel.component.amqp.AMQPConnectionDetails.AMQP_SET_TOPIC_PREFIX;
+import static
org.apache.camel.component.amqp.AMQPConnectionDetails.discoverAMQP;
+
+public class AMQPEmbeddedBrokerTest extends CamelTestSupport {
+
+ static int amqpPort = AvailablePortFinder.getNextAvailable();
+
+ static EmbeddedActiveMQ server = new EmbeddedActiveMQ();
+
+ @EndpointInject(uri = "mock:result")
+ MockEndpoint resultEndpoint;
+
+ String expectedBody = "Hello there!";
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ Configuration config = new ConfigurationImpl();
+ AddressSettings addressSettings = new AddressSettings();
+ // Disable auto create address to make sure that topic name is correct
without prefix
+ addressSettings.setAutoCreateAddresses(false);
+ config.addAcceptorConfiguration("amqp", "tcp://0.0.0.0:" + amqpPort
+ +
"?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpMinCredits=300");
+ config.setPersistenceEnabled(false);
+ config.addAddressesSetting("#", addressSettings);
+ config.setSecurityEnabled(false);
+
+ // Set explicit topic name
+ CoreAddressConfiguration pingTopicConfig = new
CoreAddressConfiguration();
+ pingTopicConfig.setName("topic.ping");
+ pingTopicConfig.addRoutingType(RoutingType.MULTICAST);
+
+ config.addAddressConfiguration(pingTopicConfig);
+
+ server.setConfiguration(config);
+ server.start();
+ System.setProperty(AMQP_PORT, amqpPort + "");
+ System.setProperty(AMQP_SET_TOPIC_PREFIX, "false");
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ server.stop();
+ }
+
+ @Test
+ public void testTopicWithoutPrefix() throws Exception {
+ resultEndpoint.expectedMessageCount(1);
+ template.sendBody("direct:send-topic", expectedBody);
+ resultEndpoint.assertIsSatisfied();
+ }
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry registry = super.createRegistry();
+ return registry;
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ JndiRegistry registry =
(JndiRegistry)((PropertyPlaceholderDelegateRegistry)camelContext.getRegistry()).getRegistry();
+ registry.bind("amqpConnection", discoverAMQP(camelContext));
+ camelContext.addComponent("amqp-customized", new AMQPComponent());
+ return camelContext;
+ }
+
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+ from("direct:send-topic")
+ .to("amqp-customized:topic:topic.ping");
+
+ from("amqp-customized:topic:topic.ping")
+ .to("log:routing")
+ .to("mock:result");
+ }
+ };
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> ActiveMQ Artemis AMQP integration issue with topic prefix hardcode
> ------------------------------------------------------------------
>
> Key: CAMEL-12624
> URL: https://issues.apache.org/jira/browse/CAMEL-12624
> Project: Camel
> Issue Type: Bug
> Components: camel-amqp
> Affects Versions: 2.21.1
> Reporter: Matt
> Priority: Minor
>
> Currently we are running Camel AMQP component against Active MQ 5 (Amazon MQ)
> we want to move to an Artemis solution but this hasn't worked seamlessly.
> In CAMEL-9204 I believe a hardcoded topic prefix of "topic://" was introduced
> I think for a workaround of an Active MQ 5 bug.
> In Artemis this means Camel connects to a topic named
> "topic://example.topic.event" instead of "example.topic.event" and therefore
> receives no events.
> The only possible workaround is to manually create the connection factory
> which means then the topic prefix is not set.
> My believe is that the hardcoded topic prefix is a bug and should be removed
> or an option to override at the AMQP component level should be introduced.
> Bug location: AMQPComponent.java line 59
> {code}
> @Override
> protected void doStart() throws Exception {
> Set<AMQPConnectionDetails> connectionDetails =
> getCamelContext().getRegistry().findByType(AMQPConnectionDetails.class);
> if (connectionDetails.size() == 1) {
> AMQPConnectionDetails details = connectionDetails.iterator().next();
> JmsConnectionFactory connectionFactory = new
> JmsConnectionFactory(details.username(), details.password(), details.uri());
> connectionFactory.setTopicPrefix("topic://");
> setConnectionFactory(connectionFactory);
> }
> super.doStart();
> }
> {code}
> Workaround:
> {code}
> @Bean
> public ConnectionFactory amqpConnectionFactory() {
> return new JmsConnectionFactory(username, password, url);
> }
> {code}
>
>
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
