You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "essobedo (via GitHub)" <gi...@apache.org> on 2023/06/08 13:00:06 UTC
[GitHub] [camel] essobedo commented on a diff in pull request #10293: CAMEL-19429: re-add camel-activemq component
essobedo commented on code in PR #10293:
URL: https://github.com/apache/camel/pull/10293#discussion_r1222978024
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
Review Comment:
```suggestion
private final List<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
private final List<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
```
It looks like we don't call any specific methods so we can use `List`
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
Review Comment:
```suggestion
if (answer.getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
activeMQConfiguration.setBrokerURL(brokerURL);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration) {
((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
```
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
activeMQConfiguration.setTrustAllPackages(trustAllPackages);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
return activeMQConfiguration.isTrustAllPackages();
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
+ }
+ }
+
+ public boolean isUsePooledConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUsePooledConnection();
+ }
+ return true;
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ @Metadata(defaultValue = "false", label = "common")
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUseSingleConnection(useSingleConnection);
+ }
+ }
+
+ public boolean isUseSingleConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUseSingleConnection();
+ }
+ return false;
+ }
+
+ @Override
+ protected void setProperties(Endpoint bean, Map<String, Object> parameters) throws Exception {
+ Object useSingleConnection = parameters.remove("useSingleConnection");
+ if (useSingleConnection != null) {
+ ((ActiveMQConfiguration) ((JmsEndpoint) bean).getConfiguration())
+ .setUseSingleConnection(
+ PropertyConfigurerSupport.property(getCamelContext(), boolean.class, useSingleConnection));
+ }
+ Object usePooledConnection = parameters.remove("usePooledConnection");
+ if (usePooledConnection != null) {
+ ((ActiveMQConfiguration) ((JmsEndpoint) bean).getConfiguration())
+ .setUsePooledConnection(
+ PropertyConfigurerSupport.property(getCamelContext(), boolean.class, usePooledConnection));
+ }
+ super.setProperties(bean, parameters);
+ }
+
+ protected void addPooledConnectionFactoryService(Service pooledConnectionFactoryService) {
+ pooledConnectionFactoryServiceList.add(pooledConnectionFactoryService);
+ }
+
+ protected void addSingleConnectionFactory(SingleConnectionFactory singleConnectionFactory) {
+ singleConnectionFactoryList.add(singleConnectionFactory);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected String convertPathToActualDestination(String path, Map<String, Object> parameters) {
+ // support ActiveMQ destination options using the destination. prefix
+ // http://activemq.apache.org/destination-options.html
+ Map options = PropertiesHelper.extractProperties(parameters, "destination.");
+
+ String query = URISupport.createQueryString(options);
+
+ // if we have destination options then append them to the destination
+ // name
+ if (ObjectHelper.isNotEmpty(query)) {
+ return path + "?" + query;
+ } else {
+ return path;
+ }
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+
+ // use OriginalDestinationPropagateStrategy by default if no custom
+ // strategy has been set
+ if (getMessageCreatedStrategy() == null) {
+ setMessageCreatedStrategy(new OriginalDestinationPropagateStrategy());
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ for (Service s : pooledConnectionFactoryServiceList) {
+ try {
+ s.stop();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ pooledConnectionFactoryServiceList.clear();
+
+ for (SingleConnectionFactory s : singleConnectionFactoryList) {
+ try {
+ s.destroy();
+ } catch (Exception e) {
+ // ignore
Review Comment:
ditto
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
+ }
+ }
+
+ public boolean isUsePooledConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUsePooledConnection();
+ }
+ return true;
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ @Metadata(defaultValue = "false", label = "common")
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUseSingleConnection(useSingleConnection);
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
activeMQConfiguration.setUseSingleConnection(useSingleConnection);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/OriginalDestinationPropagateStrategy.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.activemq;
+
+import jakarta.jms.Message;
+import jakarta.jms.Session;
+
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.jms.JmsMessage;
+import org.apache.camel.component.jms.MessageCreatedStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A strategy to enrich JMS message with their original destination if the Camel route originates from a JMS
+ * destination.
+ */
+public class OriginalDestinationPropagateStrategy implements MessageCreatedStrategy {
+
+ private static final transient Logger LOG = LoggerFactory.getLogger(OriginalDestinationPropagateStrategy.class);
+
+ @Override
+ public void onMessageCreated(Message message, Session session, Exchange exchange, Throwable cause) {
+ if (exchange.getIn() instanceof JmsMessage) {
+ JmsMessage msg = exchange.getIn(JmsMessage.class);
+ Message jms = msg.getJmsMessage();
+ if (jms != null && jms instanceof ActiveMQMessage && message instanceof ActiveMQMessage) {
+ ActiveMQMessage amq = (ActiveMQMessage) jms;
+ if (amq.getOriginalDestination() == null) {
+ ActiveMQDestination from = amq.getDestination();
+ if (from != null) {
+ LOG.trace("Setting OriginalDestination: {} on {}", from, message);
+ ((ActiveMQMessage) message).setOriginalDestination(from);
Review Comment:
```suggestion
if (exchange.getIn() instanceof JmsMessage msg) {
Message jms = msg.getJmsMessage();
if (jms != null && jms instanceof ActiveMQMessage amq && message instanceof ActiveMQMessage activeMQMessage) {
if (amq.getOriginalDestination() == null) {
ActiveMQDestination from = amq.getDestination();
if (from != null) {
LOG.trace("Setting OriginalDestination: {} on {}", from, message);
activeMQMessage.setOriginalDestination(from);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
Review Comment:
I mean this test should always pass or fail
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
return activeMQConfiguration.getBrokerURL();
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
activeMQConfiguration.setUsePooledConnection(usePooledConnection);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
activeMQConfiguration.setBrokerURL(brokerURL);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
+ }
+ }
+
+ public boolean isUsePooledConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUsePooledConnection();
+ }
+ return true;
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ @Metadata(defaultValue = "false", label = "common")
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUseSingleConnection(useSingleConnection);
+ }
+ }
+
+ public boolean isUseSingleConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUseSingleConnection();
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
return activeMQConfiguration.isUseSingleConnection();
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
+ }
+ }
+
+ public boolean isUsePooledConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUsePooledConnection();
+ }
+ return true;
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ @Metadata(defaultValue = "false", label = "common")
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUseSingleConnection(useSingleConnection);
+ }
+ }
+
+ public boolean isUseSingleConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUseSingleConnection();
+ }
+ return false;
+ }
+
+ @Override
+ protected void setProperties(Endpoint bean, Map<String, Object> parameters) throws Exception {
+ Object useSingleConnection = parameters.remove("useSingleConnection");
+ if (useSingleConnection != null) {
+ ((ActiveMQConfiguration) ((JmsEndpoint) bean).getConfiguration())
+ .setUseSingleConnection(
+ PropertyConfigurerSupport.property(getCamelContext(), boolean.class, useSingleConnection));
+ }
+ Object usePooledConnection = parameters.remove("usePooledConnection");
+ if (usePooledConnection != null) {
+ ((ActiveMQConfiguration) ((JmsEndpoint) bean).getConfiguration())
+ .setUsePooledConnection(
+ PropertyConfigurerSupport.property(getCamelContext(), boolean.class, usePooledConnection));
+ }
+ super.setProperties(bean, parameters);
+ }
+
+ protected void addPooledConnectionFactoryService(Service pooledConnectionFactoryService) {
+ pooledConnectionFactoryServiceList.add(pooledConnectionFactoryService);
+ }
+
+ protected void addSingleConnectionFactory(SingleConnectionFactory singleConnectionFactory) {
+ singleConnectionFactoryList.add(singleConnectionFactory);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected String convertPathToActualDestination(String path, Map<String, Object> parameters) {
+ // support ActiveMQ destination options using the destination. prefix
+ // http://activemq.apache.org/destination-options.html
+ Map options = PropertiesHelper.extractProperties(parameters, "destination.");
+
+ String query = URISupport.createQueryString(options);
+
+ // if we have destination options then append them to the destination
+ // name
+ if (ObjectHelper.isNotEmpty(query)) {
+ return path + "?" + query;
+ } else {
+ return path;
+ }
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+
+ // use OriginalDestinationPropagateStrategy by default if no custom
+ // strategy has been set
+ if (getMessageCreatedStrategy() == null) {
+ setMessageCreatedStrategy(new OriginalDestinationPropagateStrategy());
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ for (Service s : pooledConnectionFactoryServiceList) {
+ try {
+ s.stop();
+ } catch (Exception e) {
+ // ignore
Review Comment:
I would personally rather prefer to add a logger to this class and log the errors
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
+ }
+ }
+
+ public boolean isUsePooledConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUsePooledConnection();
Review Comment:
```suggestion
if (getConfiguration() instanceof ActiveMQConfiguration activeMQConfiguration) {
return activeMQConfiguration.isUsePooledConnection();
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQConfiguration.java:
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.activemq;
+
+import java.lang.reflect.Constructor;
+
+import jakarta.jms.ConnectionFactory;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.Service;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.connection.DelegatingConnectionFactory;
+import org.springframework.jms.connection.JmsTransactionManager;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.transaction.PlatformTransactionManager;
+
+public class ActiveMQConfiguration extends JmsConfiguration {
+ private ActiveMQComponent activeMQComponent;
+ private String brokerURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
+ private volatile boolean customBrokerURL;
+ private boolean useSingleConnection;
+ private boolean usePooledConnection = true;
+ private boolean trustAllPackages;
+
+ public ActiveMQConfiguration() {
+ }
+
+ public String getBrokerURL() {
+ return brokerURL;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ broker.
+ */
+ public void setBrokerURL(String brokerURL) {
+ this.brokerURL = brokerURL;
+ this.customBrokerURL = true;
+ }
+
+ public boolean isUseSingleConnection() {
+ return useSingleConnection;
+ }
+
+ /**
+ * @deprecated - use JmsConfiguration#getUsername()
+ * @see JmsConfiguration#getUsername()
+ */
+ @Deprecated
+ public String getUserName() {
+ return getUsername();
+ }
+
+ /**
+ * @deprecated - use JmsConfiguration#setUsername(String)
+ * @see JmsConfiguration#setUsername(String)
+ */
+ @Deprecated
+ public void setUserName(String userName) {
+ setUsername(userName);
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ this.useSingleConnection = useSingleConnection;
+ }
+
+ public boolean isUsePooledConnection() {
+ return usePooledConnection;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ this.usePooledConnection = usePooledConnection;
+ }
+
+ public boolean isTrustAllPackages() {
+ return trustAllPackages;
+ }
+
+ /**
+ * ObjectMessage objects depend on Java serialization of marshal/unmarshal object payload. This process is generally
+ * considered unsafe as malicious payload can exploit the host system. That's why starting with versions 5.12.2 and
+ * 5.13.0, ActiveMQ enforces users to explicitly whitelist packages that can be exchanged using ObjectMessages.
+ * <br/>
+ * This option can be set to <tt>true</tt> to trust all packages (eg whitelist is *).
+ * <p/>
+ * See more details at: http://activemq.apache.org/objectmessage.html
+ */
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ this.trustAllPackages = trustAllPackages;
+ }
+
+ /**
+ * Factory method to create a default transaction manager if one is not specified
+ */
+ @Override
+ protected PlatformTransactionManager createTransactionManager() {
+ JmsTransactionManager answer = new JmsTransactionManager(getOrCreateConnectionFactory());
+ answer.afterPropertiesSet();
+ return answer;
+ }
+
+ protected void setActiveMQComponent(ActiveMQComponent activeMQComponent) {
+ this.activeMQComponent = activeMQComponent;
+ }
+
+ @Override
+ public void setConnectionFactory(ConnectionFactory connectionFactory) {
+ ActiveMQConnectionFactory acf = null;
+
+ ConnectionFactory target = connectionFactory;
+ if (target instanceof CachingConnectionFactory) {
+ CachingConnectionFactory ccf = (CachingConnectionFactory) target;
+ target = ccf.getTargetConnectionFactory();
+ }
+ if (target instanceof DelegatingConnectionFactory) {
+ DelegatingConnectionFactory dcf = (DelegatingConnectionFactory) target;
+ target = dcf.getTargetConnectionFactory();
Review Comment:
```suggestion
if (target instanceof DelegatingConnectionFactory dcf) {
target = dcf.getTargetConnectionFactory();
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/converter/ActiveMQMessageConverter.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.activemq.converter;
+
+import java.io.Serializable;
+
+import jakarta.jms.JMSException;
+import jakarta.jms.Message;
+import jakarta.jms.MessageListener;
+
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQObjectMessage;
+import org.apache.activemq.command.ActiveMQTextMessage;
+import org.apache.camel.Converter;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.jms.JmsBinding;
+
+@Converter(generateLoader = true)
+public class ActiveMQMessageConverter {
+ private JmsBinding binding = new JmsBinding();
+
+ /**
+ * Converts the inbound message exchange to an ActiveMQ JMS message
+ *
+ * @return the ActiveMQ message
+ */
+ @Converter
+ public ActiveMQMessage toMessage(Exchange exchange) throws JMSException {
+ ActiveMQMessage message = createActiveMQMessage(exchange);
+ getBinding().appendJmsProperties(message, exchange);
+ return message;
+ }
+
+ /**
+ * Allows a JMS {@link MessageListener} to be converted to a Camel {@link Processor} so that we can provide better
+ * <a href="">Bean Integration</a> so that we can use any JMS MessageListener in in Camel as a bean
+ *
+ * @param listener the JMS message listener
+ * @return a newly created Camel Processor which when invoked will invoke
+ * {@link MessageListener#onMessage(Message)}
+ */
+ @Converter
+ public Processor toProcessor(final MessageListener listener) {
+ return new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ Message message = toMessage(exchange);
+ listener.onMessage(message);
+ }
+
+ @Override
+ public String toString() {
+ return "Processor of MessageListener: " + listener;
+ }
+ };
+ }
+
+ private static ActiveMQMessage createActiveMQMessage(Exchange exchange) throws JMSException {
+ Object body = exchange.getIn().getBody();
+ if (body instanceof String) {
+ ActiveMQTextMessage answer = new ActiveMQTextMessage();
+ answer.setText((String) body);
Review Comment:
```suggestion
if (body instanceof String bodyAsString) {
ActiveMQTextMessage answer = new ActiveMQTextMessage();
answer.setText(bodyAsString);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQConfiguration.java:
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.activemq;
+
+import java.lang.reflect.Constructor;
+
+import jakarta.jms.ConnectionFactory;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.Service;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.connection.DelegatingConnectionFactory;
+import org.springframework.jms.connection.JmsTransactionManager;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.transaction.PlatformTransactionManager;
+
+public class ActiveMQConfiguration extends JmsConfiguration {
+ private ActiveMQComponent activeMQComponent;
+ private String brokerURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
+ private volatile boolean customBrokerURL;
+ private boolean useSingleConnection;
+ private boolean usePooledConnection = true;
+ private boolean trustAllPackages;
+
+ public ActiveMQConfiguration() {
+ }
+
+ public String getBrokerURL() {
+ return brokerURL;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ broker.
+ */
+ public void setBrokerURL(String brokerURL) {
+ this.brokerURL = brokerURL;
+ this.customBrokerURL = true;
+ }
+
+ public boolean isUseSingleConnection() {
+ return useSingleConnection;
+ }
+
+ /**
+ * @deprecated - use JmsConfiguration#getUsername()
+ * @see JmsConfiguration#getUsername()
+ */
+ @Deprecated
+ public String getUserName() {
+ return getUsername();
+ }
+
+ /**
+ * @deprecated - use JmsConfiguration#setUsername(String)
+ * @see JmsConfiguration#setUsername(String)
+ */
+ @Deprecated
+ public void setUserName(String userName) {
+ setUsername(userName);
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ this.useSingleConnection = useSingleConnection;
+ }
+
+ public boolean isUsePooledConnection() {
+ return usePooledConnection;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ this.usePooledConnection = usePooledConnection;
+ }
+
+ public boolean isTrustAllPackages() {
+ return trustAllPackages;
+ }
+
+ /**
+ * ObjectMessage objects depend on Java serialization of marshal/unmarshal object payload. This process is generally
+ * considered unsafe as malicious payload can exploit the host system. That's why starting with versions 5.12.2 and
+ * 5.13.0, ActiveMQ enforces users to explicitly whitelist packages that can be exchanged using ObjectMessages.
+ * <br/>
+ * This option can be set to <tt>true</tt> to trust all packages (eg whitelist is *).
+ * <p/>
+ * See more details at: http://activemq.apache.org/objectmessage.html
+ */
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ this.trustAllPackages = trustAllPackages;
+ }
+
+ /**
+ * Factory method to create a default transaction manager if one is not specified
+ */
+ @Override
+ protected PlatformTransactionManager createTransactionManager() {
+ JmsTransactionManager answer = new JmsTransactionManager(getOrCreateConnectionFactory());
+ answer.afterPropertiesSet();
+ return answer;
+ }
+
+ protected void setActiveMQComponent(ActiveMQComponent activeMQComponent) {
+ this.activeMQComponent = activeMQComponent;
+ }
+
+ @Override
+ public void setConnectionFactory(ConnectionFactory connectionFactory) {
+ ActiveMQConnectionFactory acf = null;
+
+ ConnectionFactory target = connectionFactory;
+ if (target instanceof CachingConnectionFactory) {
+ CachingConnectionFactory ccf = (CachingConnectionFactory) target;
+ target = ccf.getTargetConnectionFactory();
Review Comment:
```suggestion
if (target instanceof CachingConnectionFactory ccf) {
target = ccf.getTargetConnectionFactory();
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/converter/ActiveMQMessageConverter.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.activemq.converter;
+
+import java.io.Serializable;
+
+import jakarta.jms.JMSException;
+import jakarta.jms.Message;
+import jakarta.jms.MessageListener;
+
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQObjectMessage;
+import org.apache.activemq.command.ActiveMQTextMessage;
+import org.apache.camel.Converter;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.jms.JmsBinding;
+
+@Converter(generateLoader = true)
+public class ActiveMQMessageConverter {
+ private JmsBinding binding = new JmsBinding();
+
+ /**
+ * Converts the inbound message exchange to an ActiveMQ JMS message
+ *
+ * @return the ActiveMQ message
+ */
+ @Converter
+ public ActiveMQMessage toMessage(Exchange exchange) throws JMSException {
+ ActiveMQMessage message = createActiveMQMessage(exchange);
+ getBinding().appendJmsProperties(message, exchange);
+ return message;
+ }
+
+ /**
+ * Allows a JMS {@link MessageListener} to be converted to a Camel {@link Processor} so that we can provide better
+ * <a href="">Bean Integration</a> so that we can use any JMS MessageListener in in Camel as a bean
+ *
+ * @param listener the JMS message listener
+ * @return a newly created Camel Processor which when invoked will invoke
+ * {@link MessageListener#onMessage(Message)}
+ */
+ @Converter
+ public Processor toProcessor(final MessageListener listener) {
+ return new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ Message message = toMessage(exchange);
+ listener.onMessage(message);
+ }
+
+ @Override
+ public String toString() {
+ return "Processor of MessageListener: " + listener;
+ }
+ };
+ }
+
+ private static ActiveMQMessage createActiveMQMessage(Exchange exchange) throws JMSException {
+ Object body = exchange.getIn().getBody();
+ if (body instanceof String) {
+ ActiveMQTextMessage answer = new ActiveMQTextMessage();
+ answer.setText((String) body);
+ return answer;
+ } else if (body instanceof Serializable) {
+ ActiveMQObjectMessage answer = new ActiveMQObjectMessage();
+ answer.setObject((Serializable) body);
Review Comment:
```suggestion
} else if (body instanceof Serializable bodyAsObject) {
ActiveMQObjectMessage answer = new ActiveMQObjectMessage();
answer.setObject(bodyAsObject);
```
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
+ }
+
+ return answer;
+ }
+
+ public String getBrokerURL() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).getBrokerURL();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ. If none configured then localhost:61616 is used by default
+ * (however can be overridden by configuration from environment variables)
+ */
+ @Metadata(label = "common")
+ public void setBrokerURL(String brokerURL) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setBrokerURL(brokerURL);
+ }
+ }
+
+ /**
+ * Define if all Java packages are trusted or not (for Java object JMS message types). Notice its not recommended
+ * practice to send Java serialized objects over network. Setting this to true can expose security risks, so use
+ * this with care.
+ */
+ @Metadata(defaultValue = "false", label = "advanced")
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setTrustAllPackages(trustAllPackages);
+ }
+ }
+
+ public boolean isTrustAllPackages() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isTrustAllPackages();
+ }
+ return false;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ @Metadata(defaultValue = "true", label = "common")
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUsePooledConnection(usePooledConnection);
+ }
+ }
+
+ public boolean isUsePooledConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUsePooledConnection();
+ }
+ return true;
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ @Metadata(defaultValue = "false", label = "common")
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) getConfiguration()).setUseSingleConnection(useSingleConnection);
+ }
+ }
+
+ public boolean isUseSingleConnection() {
+ if (getConfiguration() instanceof ActiveMQConfiguration) {
+ return ((ActiveMQConfiguration) getConfiguration()).isUseSingleConnection();
+ }
+ return false;
+ }
+
+ @Override
+ protected void setProperties(Endpoint bean, Map<String, Object> parameters) throws Exception {
+ Object useSingleConnection = parameters.remove("useSingleConnection");
+ if (useSingleConnection != null) {
+ ((ActiveMQConfiguration) ((JmsEndpoint) bean).getConfiguration())
+ .setUseSingleConnection(
+ PropertyConfigurerSupport.property(getCamelContext(), boolean.class, useSingleConnection));
+ }
+ Object usePooledConnection = parameters.remove("usePooledConnection");
+ if (usePooledConnection != null) {
+ ((ActiveMQConfiguration) ((JmsEndpoint) bean).getConfiguration())
+ .setUsePooledConnection(
+ PropertyConfigurerSupport.property(getCamelContext(), boolean.class, usePooledConnection));
+ }
+ super.setProperties(bean, parameters);
+ }
+
+ protected void addPooledConnectionFactoryService(Service pooledConnectionFactoryService) {
+ pooledConnectionFactoryServiceList.add(pooledConnectionFactoryService);
+ }
+
+ protected void addSingleConnectionFactory(SingleConnectionFactory singleConnectionFactory) {
+ singleConnectionFactoryList.add(singleConnectionFactory);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected String convertPathToActualDestination(String path, Map<String, Object> parameters) {
+ // support ActiveMQ destination options using the destination. prefix
+ // http://activemq.apache.org/destination-options.html
+ Map options = PropertiesHelper.extractProperties(parameters, "destination.");
+
+ String query = URISupport.createQueryString(options);
+
+ // if we have destination options then append them to the destination
+ // name
+ if (ObjectHelper.isNotEmpty(query)) {
+ return path + "?" + query;
+ } else {
+ return path;
+ }
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+
+ // use OriginalDestinationPropagateStrategy by default if no custom
+ // strategy has been set
+ if (getMessageCreatedStrategy() == null) {
+ setMessageCreatedStrategy(new OriginalDestinationPropagateStrategy());
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ for (Service s : pooledConnectionFactoryServiceList) {
+ try {
+ s.stop();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ pooledConnectionFactoryServiceList.clear();
+
+ for (SingleConnectionFactory s : singleConnectionFactoryList) {
+ try {
+ s.destroy();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ singleConnectionFactoryList.clear();
+
+ super.doStop();
+ }
+
+ /**
+ * Configuration of ActiveMQ
+ */
+ @Override
+ public void setConfiguration(JmsConfiguration configuration) {
+ if (configuration instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) configuration).setActiveMQComponent(this);
Review Comment:
```suggestion
if (configuration instanceof ActiveMQConfiguration activeMQConfiguration) {
activeMQConfiguration.setActiveMQComponent(this);
```
##########
parent/pom.xml:
##########
@@ -50,6 +50,7 @@
<checkstyle.failOnViolation>false</checkstyle.failOnViolation>
<!-- dependency versions -->
<abdera-version>1.1.3</abdera-version>
+ <activemq-version>5.18.1</activemq-version>
Review Comment:
I'm wondering if we need to add it also to `camel-dependencies/pom.xml`
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQConfiguration.java:
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.activemq;
+
+import java.lang.reflect.Constructor;
+
+import jakarta.jms.ConnectionFactory;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.Service;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.connection.DelegatingConnectionFactory;
+import org.springframework.jms.connection.JmsTransactionManager;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.transaction.PlatformTransactionManager;
+
+public class ActiveMQConfiguration extends JmsConfiguration {
+ private ActiveMQComponent activeMQComponent;
+ private String brokerURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
+ private volatile boolean customBrokerURL;
+ private boolean useSingleConnection;
+ private boolean usePooledConnection = true;
+ private boolean trustAllPackages;
+
+ public ActiveMQConfiguration() {
+ }
+
+ public String getBrokerURL() {
+ return brokerURL;
+ }
+
+ /**
+ * Sets the broker URL to use to connect to ActiveMQ broker.
+ */
+ public void setBrokerURL(String brokerURL) {
+ this.brokerURL = brokerURL;
+ this.customBrokerURL = true;
+ }
+
+ public boolean isUseSingleConnection() {
+ return useSingleConnection;
+ }
+
+ /**
+ * @deprecated - use JmsConfiguration#getUsername()
+ * @see JmsConfiguration#getUsername()
+ */
+ @Deprecated
+ public String getUserName() {
+ return getUsername();
+ }
+
+ /**
+ * @deprecated - use JmsConfiguration#setUsername(String)
+ * @see JmsConfiguration#setUsername(String)
+ */
+ @Deprecated
+ public void setUserName(String userName) {
+ setUsername(userName);
+ }
+
+ /**
+ * Enables or disables whether a Spring {@link SingleConnectionFactory} will be used so that when messages are sent
+ * to ActiveMQ from outside of a message consuming thread, pooling will be used rather than the default with the
+ * Spring {@link JmsTemplate} which will create a new connection, session, producer for each message then close them
+ * all down again.
+ * <p/>
+ * The default value is false and a pooled connection is used by default.
+ */
+ public void setUseSingleConnection(boolean useSingleConnection) {
+ this.useSingleConnection = useSingleConnection;
+ }
+
+ public boolean isUsePooledConnection() {
+ return usePooledConnection;
+ }
+
+ /**
+ * Enables or disables whether a PooledConnectionFactory will be used so that when messages are sent to ActiveMQ
+ * from outside of a message consuming thread, pooling will be used rather than the default with the Spring
+ * {@link JmsTemplate} which will create a new connection, session, producer for each message then close them all
+ * down again.
+ * <p/>
+ * The default value is true.
+ */
+ public void setUsePooledConnection(boolean usePooledConnection) {
+ this.usePooledConnection = usePooledConnection;
+ }
+
+ public boolean isTrustAllPackages() {
+ return trustAllPackages;
+ }
+
+ /**
+ * ObjectMessage objects depend on Java serialization of marshal/unmarshal object payload. This process is generally
+ * considered unsafe as malicious payload can exploit the host system. That's why starting with versions 5.12.2 and
+ * 5.13.0, ActiveMQ enforces users to explicitly whitelist packages that can be exchanged using ObjectMessages.
+ * <br/>
+ * This option can be set to <tt>true</tt> to trust all packages (eg whitelist is *).
+ * <p/>
+ * See more details at: http://activemq.apache.org/objectmessage.html
+ */
+ public void setTrustAllPackages(boolean trustAllPackages) {
+ this.trustAllPackages = trustAllPackages;
+ }
+
+ /**
+ * Factory method to create a default transaction manager if one is not specified
+ */
+ @Override
+ protected PlatformTransactionManager createTransactionManager() {
+ JmsTransactionManager answer = new JmsTransactionManager(getOrCreateConnectionFactory());
+ answer.afterPropertiesSet();
+ return answer;
+ }
+
+ protected void setActiveMQComponent(ActiveMQComponent activeMQComponent) {
+ this.activeMQComponent = activeMQComponent;
+ }
+
+ @Override
+ public void setConnectionFactory(ConnectionFactory connectionFactory) {
+ ActiveMQConnectionFactory acf = null;
+
+ ConnectionFactory target = connectionFactory;
+ if (target instanceof CachingConnectionFactory) {
+ CachingConnectionFactory ccf = (CachingConnectionFactory) target;
+ target = ccf.getTargetConnectionFactory();
+ }
+ if (target instanceof DelegatingConnectionFactory) {
+ DelegatingConnectionFactory dcf = (DelegatingConnectionFactory) target;
+ target = dcf.getTargetConnectionFactory();
+ }
+ if (target instanceof ActiveMQConnectionFactory) {
+ acf = (ActiveMQConnectionFactory) target;
+ }
Review Comment:
Would it make sense to use if / else if blocks instead in this case?
##########
components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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.activemq;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.activemq.Service;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.component.jms.JmsConfiguration;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.jms.QueueBrowseStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
+import org.apache.camel.util.URISupport;
+import org.springframework.jms.connection.SingleConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * The ActiveMQ Component.
+ */
+@Component("activemq")
+public class ActiveMQComponent extends JmsComponent {
+ private final CopyOnWriteArrayList<SingleConnectionFactory> singleConnectionFactoryList = new CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Service> pooledConnectionFactoryServiceList = new CopyOnWriteArrayList<>();
+
+ public ActiveMQComponent() {
+ }
+
+ public ActiveMQComponent(CamelContext context) {
+ super(context);
+ }
+
+ public ActiveMQComponent(ActiveMQConfiguration configuration) {
+ setConfiguration(configuration);
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a>
+ *
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent() {
+ return new ActiveMQComponent();
+ }
+
+ /**
+ * Creates an <a href="http://camel.apache.org/activemq.html">ActiveMQ Component</a> connecting to the given
+ * <a href="http://activemq.apache.org/configuring-transports.html">broker URL</a>
+ *
+ * @param brokerURL the URL to connect to
+ * @return the created component
+ */
+ public static ActiveMQComponent activeMQComponent(String brokerURL) {
+ ActiveMQComponent answer = new ActiveMQComponent();
+ if (answer.getConfiguration() instanceof ActiveMQConfiguration) {
+ ((ActiveMQConfiguration) answer.getConfiguration()).setBrokerURL(brokerURL);
Review Comment:
In which case can we have another type?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org