This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new ab32d49ea152 CAMEL-26693: camel-activemq, camel-activemq6 - endpoints
keep working after the component is stopped and started (#29344)
ab32d49ea152 is described below
commit ab32d49ea15215fbca4d2215fd981b29a2007b4e
Author: Federico Mariani <[email protected]>
AuthorDate: Sat Nov 7 15:49:46 2026 +0100
CAMEL-26693: camel-activemq, camel-activemq6 - endpoints keep working after
the component is stopped and started (#29344)
---
.../camel/catalog/docs/activemq-component.adoc | 26 +++
.../camel/catalog/docs/activemq6-component.adoc | 26 +++
components/camel-activemq/pom.xml | 7 +
.../src/main/docs/activemq-component.adoc | 26 +++
.../component/activemq/ActiveMQComponent.java | 36 +++-
.../activemq/ActiveMQComponentRestartTest.java | 75 ++++++++
components/camel-activemq6/pom.xml | 13 ++
.../src/main/docs/activemq6-component.adoc | 26 +++
.../component/activemq6/ActiveMQComponent.java | 36 +++-
.../activemq6/ActiveMQComponentRestartTest.java | 192 +++++++++++++++++++++
10 files changed, 451 insertions(+), 12 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq-component.adoc
index c1adbaaebe5e..459da0fd9768 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq-component.adoc
@@ -142,3 +142,29 @@ YAML::
message: "Received a message - ${body}"
----
====
+
+== Connection factory and transactions
+
+With only `brokerURL` set, each endpoint creates its own connection factory,
and with `usePooledConnection=true` (the
+default) its own pool, which needs `org.messaginghub:pooled-jms` on the
classpath. A message sent from a transacted
+route, for example
`from("activemq:queue:in?transacted=true").to("activemq:queue:out")`, is then
not part of the
+transaction of the consumed message: when the route fails, the consumed
message is rolled back and delivered again,
+but the message sent to `out` is not rolled back.
+
+To send in the transaction of the consumer, set one `connectionFactory` on the
component, for example a pooled one,
+so that the endpoints share it. A `transactionManager` that you set must use
that same connection factory: a
+transaction manager on another connection factory does not make the message
part of the transaction.
+
+[source,properties]
+----
+camel.beans.activemqCF = #class:org.apache.activemq.ActiveMQConnectionFactory
+camel.beans.activemqCF.brokerURL = tcp://localhost:61616
+camel.beans.pooledCF =
#class:org.messaginghub.pooled.jms.JmsPoolConnectionFactory
+camel.beans.pooledCF.connectionFactory = #bean:activemqCF
+camel.component.activemq.connection-factory = #bean:pooledCF
+----
+
+When the endpoints share one pool, they get the same connection, whose client
id cannot be changed: give an endpoint
+with its own `clientId` (for example for a durable subscription) a
`connectionFactory` of its own, otherwise its
+consumer cannot connect and keeps retrying. `useSingleConnection=true` cannot
be combined with a `clientId`: the shared
+connection of the Spring `SingleConnectionFactory` refuses to set a client id.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq6-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq6-component.adoc
index d05cbdce1a20..392b9b66c892 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq6-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/activemq6-component.adoc
@@ -142,3 +142,29 @@ YAML::
message: "Received a message - ${body}"
----
====
+
+== Connection factory and transactions
+
+With only `brokerURL` set, each endpoint creates its own connection factory,
and with `usePooledConnection=true` (the
+default) its own pool, which needs `org.messaginghub:pooled-jms` on the
classpath. A message sent from a transacted
+route, for example
`from("activemq6:queue:in?transacted=true").to("activemq6:queue:out")`, is then
not part of the
+transaction of the consumed message: when the route fails, the consumed
message is rolled back and delivered again,
+but the message sent to `out` is not rolled back.
+
+To send in the transaction of the consumer, set one `connectionFactory` on the
component, for example a pooled one,
+so that the endpoints share it. A `transactionManager` that you set must use
that same connection factory: a
+transaction manager on another connection factory does not make the message
part of the transaction.
+
+[source,properties]
+----
+camel.beans.activemqCF = #class:org.apache.activemq.ActiveMQConnectionFactory
+camel.beans.activemqCF.brokerURL = tcp://localhost:61616
+camel.beans.pooledCF =
#class:org.messaginghub.pooled.jms.JmsPoolConnectionFactory
+camel.beans.pooledCF.connectionFactory = #bean:activemqCF
+camel.component.activemq6.connection-factory = #bean:pooledCF
+----
+
+When the endpoints share one pool, they get the same connection, whose client
id cannot be changed: give an endpoint
+with its own `clientId` (for example for a durable subscription) a
`connectionFactory` of its own, otherwise its
+consumer cannot connect and keeps retrying. `useSingleConnection=true` cannot
be combined with a `clientId`: the shared
+connection of the Spring `SingleConnectionFactory` refuses to set a client id.
diff --git a/components/camel-activemq/pom.xml
b/components/camel-activemq/pom.xml
index d758fb0dbca8..2f85ab3f96d4 100644
--- a/components/camel-activemq/pom.xml
+++ b/components/camel-activemq/pom.xml
@@ -69,6 +69,13 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <!-- the pool the component uses by default, for the unit tests -->
+ <dependency>
+ <groupId>org.messaginghub</groupId>
+ <artifactId>pooled-jms</artifactId>
+ <version>${pooled-jms-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/components/camel-activemq/src/main/docs/activemq-component.adoc
b/components/camel-activemq/src/main/docs/activemq-component.adoc
index c1adbaaebe5e..459da0fd9768 100644
--- a/components/camel-activemq/src/main/docs/activemq-component.adoc
+++ b/components/camel-activemq/src/main/docs/activemq-component.adoc
@@ -142,3 +142,29 @@ YAML::
message: "Received a message - ${body}"
----
====
+
+== Connection factory and transactions
+
+With only `brokerURL` set, each endpoint creates its own connection factory,
and with `usePooledConnection=true` (the
+default) its own pool, which needs `org.messaginghub:pooled-jms` on the
classpath. A message sent from a transacted
+route, for example
`from("activemq:queue:in?transacted=true").to("activemq:queue:out")`, is then
not part of the
+transaction of the consumed message: when the route fails, the consumed
message is rolled back and delivered again,
+but the message sent to `out` is not rolled back.
+
+To send in the transaction of the consumer, set one `connectionFactory` on the
component, for example a pooled one,
+so that the endpoints share it. A `transactionManager` that you set must use
that same connection factory: a
+transaction manager on another connection factory does not make the message
part of the transaction.
+
+[source,properties]
+----
+camel.beans.activemqCF = #class:org.apache.activemq.ActiveMQConnectionFactory
+camel.beans.activemqCF.brokerURL = tcp://localhost:61616
+camel.beans.pooledCF =
#class:org.messaginghub.pooled.jms.JmsPoolConnectionFactory
+camel.beans.pooledCF.connectionFactory = #bean:activemqCF
+camel.component.activemq.connection-factory = #bean:pooledCF
+----
+
+When the endpoints share one pool, they get the same connection, whose client
id cannot be changed: give an endpoint
+with its own `clientId` (for example for a durable subscription) a
`connectionFactory` of its own, otherwise its
+consumer cannot connect and keeps retrying. `useSingleConnection=true` cannot
be combined with a `clientId`: the shared
+connection of the Spring `SingleConnectionFactory` refuses to set a client id.
diff --git
a/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java
b/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java
index cadd020ab33e..520e03f6859b 100644
---
a/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java
+++
b/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ActiveMQComponent.java
@@ -238,27 +238,51 @@ public class ActiveMQComponent extends JmsComponent {
@Override
protected void doStop() throws Exception {
+ // close the connections of the connection factories that the
endpoints created, but keep the factories usable:
+ // the endpoints keep them, and open new connections when the
component is started again
for (Object s : pooledConnectionFactoryServiceList) {
+ invokePooledConnectionFactory(s, "clear");
+ }
+ for (SingleConnectionFactory s : singleConnectionFactoryList) {
try {
- // invoke stop method if exists
- Method m = s.getClass().getMethod("stop");
- org.apache.camel.support.ObjectHelper.invokeMethod(m, s);
+ s.resetConnection();
} catch (Exception e) {
- // ignore
+ LOG.debug("Error resetting the connection of {}. This
exception is ignored.", s, e);
}
}
+
+ super.doStop();
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ for (Object s : pooledConnectionFactoryServiceList) {
+ invokePooledConnectionFactory(s, "stop");
+ }
pooledConnectionFactoryServiceList.clear();
for (SingleConnectionFactory s : singleConnectionFactoryList) {
try {
s.destroy();
} catch (Exception e) {
- // ignore
+ LOG.debug("Error destroying {}. This exception is ignored.",
s, e);
}
}
singleConnectionFactoryList.clear();
- super.doStop();
+ super.doShutdown();
+ }
+
+ private static void invokePooledConnectionFactory(Object
pooledConnectionFactory, String method) {
+ try {
+ // invoke the method if it exists
+ Method m = pooledConnectionFactory.getClass().getMethod(method);
+ org.apache.camel.support.ObjectHelper.invokeMethod(m,
pooledConnectionFactory);
+ } catch (NoSuchMethodException e) {
+ // the connection factory does not have this method
+ } catch (Exception e) {
+ LOG.debug("Error invoking {} on {}. This exception is ignored.",
method, pooledConnectionFactory, e);
+ }
}
/**
diff --git
a/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/ActiveMQComponentRestartTest.java
b/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/ActiveMQComponentRestartTest.java
new file mode 100644
index 000000000000..45a22e20a5fe
--- /dev/null
+++
b/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/ActiveMQComponentRestartTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.JMSException;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * The pool that the component creates for an endpoint from the broker URL
stays usable when the component is stopped
+ * and started again, and is stopped when Camel stops. No broker listens on
the broker URL: a usable pool tries to
+ * connect and fails, a stopped pool gives no connection.
+ */
+public class ActiveMQComponentRestartTest {
+
+ private static JmsPoolConnectionFactory pool(CamelContext context) {
+ JmsEndpoint endpoint = context.getEndpoint("activemq:queue:restart",
JmsEndpoint.class);
+ return assertInstanceOf(JmsPoolConnectionFactory.class,
endpoint.getConfiguration().getOrCreateConnectionFactory());
+ }
+
+ private static CamelContext context() {
+ ActiveMQComponent activemq = new ActiveMQComponent();
+ activemq.setBrokerURL("tcp://localhost:1");
+ CamelContext context = new DefaultCamelContext();
+ context.addComponent("activemq", activemq);
+ return context;
+ }
+
+ @Test
+ public void poolIsUsableAfterTheComponentIsStoppedAndStarted() throws
Exception {
+ try (CamelContext context = context()) {
+ context.start();
+ JmsPoolConnectionFactory pool = pool(context);
+
+ ActiveMQComponent activemq = context.getComponent("activemq",
ActiveMQComponent.class);
+ activemq.stop();
+ activemq.start();
+
+ assertThrows(JMSException.class, pool::createConnection);
+ }
+ }
+
+ @Test
+ public void poolIsStoppedWhenTheContextStops() throws Exception {
+ JmsPoolConnectionFactory pool;
+ try (CamelContext context = context()) {
+ context.start();
+ pool = pool(context);
+ }
+
+ assertNull(pool.createConnection());
+ }
+}
diff --git a/components/camel-activemq6/pom.xml
b/components/camel-activemq6/pom.xml
index 19c966347cc2..c82f013b954b 100644
--- a/components/camel-activemq6/pom.xml
+++ b/components/camel-activemq6/pom.xml
@@ -70,6 +70,19 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <!-- an embedded broker, and the pool the component uses by default,
for the unit tests -->
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-broker</artifactId>
+ <version>${activemq6-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.messaginghub</groupId>
+ <artifactId>pooled-jms</artifactId>
+ <version>${pooled-jms-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/components/camel-activemq6/src/main/docs/activemq6-component.adoc
b/components/camel-activemq6/src/main/docs/activemq6-component.adoc
index d05cbdce1a20..392b9b66c892 100644
--- a/components/camel-activemq6/src/main/docs/activemq6-component.adoc
+++ b/components/camel-activemq6/src/main/docs/activemq6-component.adoc
@@ -142,3 +142,29 @@ YAML::
message: "Received a message - ${body}"
----
====
+
+== Connection factory and transactions
+
+With only `brokerURL` set, each endpoint creates its own connection factory,
and with `usePooledConnection=true` (the
+default) its own pool, which needs `org.messaginghub:pooled-jms` on the
classpath. A message sent from a transacted
+route, for example
`from("activemq6:queue:in?transacted=true").to("activemq6:queue:out")`, is then
not part of the
+transaction of the consumed message: when the route fails, the consumed
message is rolled back and delivered again,
+but the message sent to `out` is not rolled back.
+
+To send in the transaction of the consumer, set one `connectionFactory` on the
component, for example a pooled one,
+so that the endpoints share it. A `transactionManager` that you set must use
that same connection factory: a
+transaction manager on another connection factory does not make the message
part of the transaction.
+
+[source,properties]
+----
+camel.beans.activemqCF = #class:org.apache.activemq.ActiveMQConnectionFactory
+camel.beans.activemqCF.brokerURL = tcp://localhost:61616
+camel.beans.pooledCF =
#class:org.messaginghub.pooled.jms.JmsPoolConnectionFactory
+camel.beans.pooledCF.connectionFactory = #bean:activemqCF
+camel.component.activemq6.connection-factory = #bean:pooledCF
+----
+
+When the endpoints share one pool, they get the same connection, whose client
id cannot be changed: give an endpoint
+with its own `clientId` (for example for a durable subscription) a
`connectionFactory` of its own, otherwise its
+consumer cannot connect and keeps retrying. `useSingleConnection=true` cannot
be combined with a `clientId`: the shared
+connection of the Spring `SingleConnectionFactory` refuses to set a client id.
diff --git
a/components/camel-activemq6/src/main/java/org/apache/camel/component/activemq6/ActiveMQComponent.java
b/components/camel-activemq6/src/main/java/org/apache/camel/component/activemq6/ActiveMQComponent.java
index f5627a21b518..47133b9eae72 100644
---
a/components/camel-activemq6/src/main/java/org/apache/camel/component/activemq6/ActiveMQComponent.java
+++
b/components/camel-activemq6/src/main/java/org/apache/camel/component/activemq6/ActiveMQComponent.java
@@ -238,27 +238,51 @@ public class ActiveMQComponent extends JmsComponent {
@Override
protected void doStop() throws Exception {
+ // close the connections of the connection factories that the
endpoints created, but keep the factories usable:
+ // the endpoints keep them, and open new connections when the
component is started again
for (Object s : pooledConnectionFactoryServiceList) {
+ invokePooledConnectionFactory(s, "clear");
+ }
+ for (SingleConnectionFactory s : singleConnectionFactoryList) {
try {
- // invoke stop method if exists
- Method m = s.getClass().getMethod("stop");
- org.apache.camel.support.ObjectHelper.invokeMethod(m, s);
+ s.resetConnection();
} catch (Exception e) {
- // ignore
+ LOG.debug("Error resetting the connection of {}. This
exception is ignored.", s, e);
}
}
+
+ super.doStop();
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ for (Object s : pooledConnectionFactoryServiceList) {
+ invokePooledConnectionFactory(s, "stop");
+ }
pooledConnectionFactoryServiceList.clear();
for (SingleConnectionFactory s : singleConnectionFactoryList) {
try {
s.destroy();
} catch (Exception e) {
- // ignore
+ LOG.debug("Error destroying {}. This exception is ignored.",
s, e);
}
}
singleConnectionFactoryList.clear();
- super.doStop();
+ super.doShutdown();
+ }
+
+ private static void invokePooledConnectionFactory(Object
pooledConnectionFactory, String method) {
+ try {
+ // invoke the method if it exists
+ Method m = pooledConnectionFactory.getClass().getMethod(method);
+ org.apache.camel.support.ObjectHelper.invokeMethod(m,
pooledConnectionFactory);
+ } catch (NoSuchMethodException e) {
+ // the connection factory does not have this method
+ } catch (Exception e) {
+ LOG.debug("Error invoking {} on {}. This exception is ignored.",
method, pooledConnectionFactory, e);
+ }
}
/**
diff --git
a/components/camel-activemq6/src/test/java/org/apache/camel/component/activemq6/ActiveMQComponentRestartTest.java
b/components/camel-activemq6/src/test/java/org/apache/camel/component/activemq6/ActiveMQComponentRestartTest.java
new file mode 100644
index 000000000000..be299bc3e4e1
--- /dev/null
+++
b/components/camel-activemq6/src/test/java/org/apache/camel/component/activemq6/ActiveMQComponentRestartTest.java
@@ -0,0 +1,192 @@
+/*
+ * 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.activemq6;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.jms.JmsEndpoint;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The connection factories that the component creates from the broker URL
keep working when the component, a route or
+ * the CamelContext is stopped and started again, and their connections are
closed when Camel stops.
+ */
+public class ActiveMQComponentRestartTest {
+
+ private BrokerService broker;
+ private String brokerUrl;
+
+ @BeforeEach
+ void startBroker() throws Exception {
+ broker = new BrokerService();
+ broker.setBrokerName("restart");
+ broker.setPersistent(false);
+ broker.setUseJmx(true);
+ broker.getManagementContext().setCreateConnector(false);
+ broker.setAdvisorySupport(false);
+ TransportConnector connector =
broker.addConnector("tcp://localhost:0");
+ broker.start();
+ broker.waitUntilStarted();
+ brokerUrl = connector.getConnectUri().toString();
+ }
+
+ @AfterEach
+ void stopBroker() throws Exception {
+ broker.stop();
+ }
+
+ private long brokerConnections() throws Exception {
+ return broker.getAdminView().getCurrentConnectionsCount();
+ }
+
+ private ActiveMQComponent activemq(boolean singleConnection) {
+ ActiveMQComponent activemq = new ActiveMQComponent();
+ activemq.setBrokerURL(brokerUrl);
+ activemq.setUseSingleConnection(singleConnection);
+ return activemq;
+ }
+
+ private static CamelContext context(ActiveMQComponent activemq, String
queue) throws Exception {
+ CamelContext context = new DefaultCamelContext();
+ context.addComponent("activemq", activemq);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("activemq:queue:" +
queue).routeId("consumer").to("mock:result");
+ }
+ });
+ return context;
+ }
+
+ private static void assertSendAndReceive(CamelContext context, String
queue, String body) throws Exception {
+ MockEndpoint result = context.getEndpoint("mock:result",
MockEndpoint.class);
+ result.reset();
+ result.expectedBodiesReceived(body);
+ context.createProducerTemplate().sendBody("activemq:queue:" + queue,
body);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ private void awaitNoBrokerConnections() {
+ await().atMost(10, TimeUnit.SECONDS).until(() -> brokerConnections()
== 0);
+ }
+
+ @Test
+ public void componentStopAndStartWithPooledConnection() throws Exception {
+ componentStopAndStart(false, "restart.component.pooled");
+ }
+
+ @Test
+ public void componentStopAndStartWithSingleConnection() throws Exception {
+ componentStopAndStart(true, "restart.component.single");
+ }
+
+ private void componentStopAndStart(boolean singleConnection, String queue)
throws Exception {
+ ActiveMQComponent activemq = activemq(singleConnection);
+ CamelContext context = context(activemq, queue);
+ context.start();
+ try {
+ assertSendAndReceive(context, queue, "before");
+
+ context.getRouteController().stopRoute("consumer");
+ activemq.stop();
+ awaitNoBrokerConnections();
+ activemq.start();
+ context.getRouteController().startRoute("consumer");
+
+ assertSendAndReceive(context, queue, "after");
+
+ // and the connections opened after the restart are closed when
the component stops again
+ context.getRouteController().stopRoute("consumer");
+ activemq.stop();
+ awaitNoBrokerConnections();
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void routeStopAndStart() throws Exception {
+ String queue = "restart.route";
+ CamelContext context = context(activemq(false), queue);
+ context.start();
+ try {
+ assertSendAndReceive(context, queue, "before");
+
+ context.getRouteController().stopRoute("consumer");
+ context.getRouteController().startRoute("consumer");
+
+ assertSendAndReceive(context, queue, "after");
+ } finally {
+ context.stop();
+ }
+ awaitNoBrokerConnections();
+ }
+
+ @Test
+ public void contextStopAndStart() throws Exception {
+ String queue = "restart.context";
+ ActiveMQComponent activemq = activemq(false);
+ CamelContext context = context(activemq, queue);
+ context.start();
+ try {
+ assertSendAndReceive(context, queue, "before");
+
+ // the CamelContext shuts its components down when it stops, and
creates its endpoints again when it starts
+ context.stop();
+ awaitNoBrokerConnections();
+ context.addComponent("activemq", activemq);
+ context.start();
+
+ assertSendAndReceive(context, queue, "after");
+ } finally {
+ context.stop();
+ }
+ awaitNoBrokerConnections();
+ }
+
+ @Test
+ public void shutdownStopsThePools() throws Exception {
+ String queue = "restart.shutdown";
+ CamelContext context = context(activemq(false), queue);
+ context.start();
+ JmsPoolConnectionFactory pool;
+ try {
+ assertSendAndReceive(context, queue, "Hello");
+ pool = assertInstanceOf(JmsPoolConnectionFactory.class,
+ context.getEndpoint("activemq:queue:" + queue,
JmsEndpoint.class).getConfiguration()
+ .getConnectionFactory());
+ } finally {
+ context.stop();
+ }
+ awaitNoBrokerConnections();
+ // a stopped pool gives no connection
+ assertNull(pool.createConnection());
+ }
+}
