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 48ea92c4092b CAMEL-26453: camel-dynamic-router - a route that sends to
a channel can be added again after it was removed (#28810)
48ea92c4092b is described below
commit 48ea92c4092b398f8f657320c7ac050d40da6bef
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Nov 2 16:22:27 2026 +0530
CAMEL-26453: camel-dynamic-router - a route that sends to a channel can be
added again after it was removed (#28810)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../routing/DynamicRouterComponent.java | 90 ++++++-
.../routing/DynamicRouterEndpoint.java | 46 +++-
.../routing/DynamicRouterProcessor.java | 18 ++
.../routing/DynamicRouterProducer.java | 27 +-
.../routing/DynamicRouterComponentTest.java | 55 ++++-
.../routing/DynamicRouterProducerTest.java | 20 ++
.../routing/DynamicRouterRemoveRouteTest.java | 275 +++++++++++++++++++++
7 files changed, 497 insertions(+), 34 deletions(-)
diff --git
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponent.java
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponent.java
index 09113ba59a26..a8abafb0859e 100644
---
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponent.java
+++
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponent.java
@@ -16,8 +16,13 @@
*/
package org.apache.camel.component.dynamicrouter.routing;
-import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiFunction;
import java.util.function.Supplier;
@@ -63,7 +68,21 @@ public class DynamicRouterComponent extends DefaultComponent
{
/**
* The {@link DynamicRouterProcessor}s, mapped by their channel, for the
Dynamic Router.
*/
- private final Map<String, DynamicRouterProcessor> processors = new
HashMap<>();
+ private final Map<String, DynamicRouterProcessor> processors = new
ConcurrentHashMap<>();
+
+ /**
+ * The endpoint that uses the processor of each channel, mapped by their
channel.
+ */
+ private final Map<String, DynamicRouterEndpoint> processorEndpoints = new
ConcurrentHashMap<>();
+
+ /**
+ * The processors that were replaced by a processor for an endpoint with
other options. Their recipient lists are
+ * only stopped when they are replaced, as another route can still have
exchanges in flight in them, and are shut
+ * down with the component. The threads of a thread pool of the default
profile end after its keep-alive time.
+ */
+ private final List<DynamicRouterProcessor> replacedProcessors = new
CopyOnWriteArrayList<>();
+
+ private final Lock processorsLock = new ReentrantLock();
/**
* Creates a {@link DynamicRouterEndpoint} instance.
@@ -158,22 +177,69 @@ public class DynamicRouterComponent extends
DefaultComponent {
*/
@Override
protected void doShutdown() throws Exception {
- ServiceHelper.stopAndShutdownService(processors);
+ // the recipient lists of the processors of the channels and of the
replaced processors, with their thread pools
+ List<RecipientList> recipientLists = new ArrayList<>();
+ processors.values().forEach(p ->
recipientLists.add(p.getRecipientList()));
+ replacedProcessors.forEach(p ->
recipientLists.add(p.getRecipientList()));
processors.clear();
- super.doShutdown();
+ processorEndpoints.clear();
+ replacedProcessors.clear();
+ try {
+ // tries every one of them, and rethrows the first failure
+ ServiceHelper.stopAndShutdownServices(recipientLists);
+ } finally {
+ super.doShutdown();
+ }
}
/**
- * If no processor has been added for the supplied channel, then add the
supplied processor.
+ * Get the processor of the channel of the endpoint, and create it if the
channel has none.
+ * <ul>
+ * <li>An endpoint with the same URI as the endpoint that uses the
processor uses the same processor, such as when
+ * the route that used the endpoint is removed and added again (also by a
route reload), or when a {@code toD}
+ * creates the endpoint again after the endpoint was removed or
evicted.</li>
+ * <li>An endpoint with another URI for the channel fails while the
endpoint that uses the processor is not stopped,
+ * as there can be only one processor per channel. When that endpoint is
stopped (such as with the route that used
+ * it), the processor is replaced by a processor for the new endpoint, and
the recipient list of the old processor
+ * is stopped (its thread pool is only shut down with the component, so
that the exchanges that another route still
+ * has in flight in it are not interrupted or dropped).</li>
+ * </ul>
*
- * @param channel the channel to add the processor to
- * @param processor the processor to add for the channel
+ * @param endpoint the endpoint
+ * @param factory creates the processor for the endpoint
+ * @return the processor of the channel of the endpoint
*/
- void addRoutingProcessor(final String channel, final
DynamicRouterProcessor processor) {
- if (processors.putIfAbsent(channel, processor) != null) {
- throw new IllegalArgumentException(
- "Dynamic Router can have only one processor per channel;
channel '"
- + channel + "' already has a
processor");
+ DynamicRouterProcessor getOrCreateRoutingProcessor(
+ final DynamicRouterEndpoint endpoint, final
Supplier<DynamicRouterProcessor> factory) {
+ final String channel = endpoint.getConfiguration().getChannel();
+ processorsLock.lock();
+ try {
+ DynamicRouterProcessor current = processors.get(channel);
+ if (current != null) {
+ DynamicRouterEndpoint user = processorEndpoints.get(channel);
+ if (user != null && (user == endpoint ||
user.getEndpointUri().equals(endpoint.getEndpointUri()))) {
+ processorEndpoints.put(channel, endpoint);
+ return current;
+ }
+ if (user == null || !user.isProcessorReleased()) {
+ throw new IllegalArgumentException(
+ "Dynamic Router can have only one processor per
channel; channel '"
+ + channel + "' already
has a processor");
+ }
+ LOG.info("Replacing the processor of channel '{}' of the
stopped endpoint {} for the endpoint {}",
+ channel, user.getEndpointUri(),
endpoint.getEndpointUri());
+ }
+ // create the new processor first, so that the channel keeps its
processor when that fails
+ DynamicRouterProcessor created = factory.get();
+ processors.put(channel, created);
+ processorEndpoints.put(channel, endpoint);
+ if (current != null) {
+ replacedProcessors.add(current);
+ current.stopRecipientList();
+ }
+ return created;
+ } finally {
+ processorsLock.unlock();
}
}
diff --git
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterEndpoint.java
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterEndpoint.java
index cda0e5fc8b51..c891fa9850df 100644
---
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterEndpoint.java
+++
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterEndpoint.java
@@ -105,8 +105,14 @@ public class DynamicRouterEndpoint extends DefaultEndpoint
{
private final DynamicRouterFilterService filterService;
/**
- * Create the Dynamic Router {@link org.apache.camel.Endpoint} for the
given endpoint URI. This includes the
- * creation of a {@link DynamicRouterProcessor} that is registered with
the supplied {@link DynamicRouterComponent}.
+ * Whether this endpoint has been stopped (and not initialized again).
+ */
+ private volatile boolean processorReleased;
+
+ /**
+ * Create the Dynamic Router {@link org.apache.camel.Endpoint} for the
given endpoint URI. When it is initialized,
+ * the endpoint gets the {@link DynamicRouterProcessor} of its channel
from the supplied
+ * {@link DynamicRouterComponent}, which creates it when the channel has
none.
*
* @param uri the endpoint URI
* @param component the Dynamic Router {@link
org.apache.camel.Component}
@@ -160,7 +166,9 @@ public class DynamicRouterEndpoint extends DefaultEndpoint {
}
/**
- * Initialize the endpoint by creating (and adding) the {@link
DynamicRouterProcessor} instance.
+ * Initialize the endpoint by getting the {@link DynamicRouterProcessor}
of its channel from the component, which
+ * creates it with the {@link #processorFactorySupplier} when the channel
has none (see
+ * {@link DynamicRouterComponent#getOrCreateRoutingProcessor}).
*
* @throws Exception if there is a problem getting the {@link
DynamicRouterProcessor} instance through the
* {@link #processorFactorySupplier}
@@ -168,11 +176,27 @@ public class DynamicRouterEndpoint extends
DefaultEndpoint {
@Override
protected void doInit() throws Exception {
super.doInit();
- DynamicRouterComponent component = getDynamicRouterComponent();
- CamelContext camelContext = getCamelContext();
- DynamicRouterProcessor processor = processorFactorySupplier.get()
- .getInstance(camelContext, configuration, filterService,
recipientListSupplier);
- component.addRoutingProcessor(configuration.getChannel(), processor);
+ processorReleased = false;
+ getDynamicRouterComponent().getOrCreateRoutingProcessor(this, () ->
processorFactorySupplier.get()
+ .getInstance(getCamelContext(), configuration, filterService,
recipientListSupplier));
+ }
+
+ /**
+ * Marks the processor of the channel as released by this endpoint: an
endpoint with another URI for the channel can
+ * then replace it, such as when the route that used this endpoint is
removed and the route is added again with
+ * other options. An endpoint with the same URI uses the same processor
again.
+ */
+ @Override
+ protected void doStop() throws Exception {
+ processorReleased = true;
+ super.doStop();
+ }
+
+ /**
+ * Whether this endpoint has been stopped, so that the processor of its
channel can be replaced.
+ */
+ boolean isProcessorReleased() {
+ return processorReleased;
}
/**
@@ -220,9 +244,9 @@ public class DynamicRouterEndpoint extends DefaultEndpoint {
public static class DynamicRouterEndpointFactory {
/**
- * Create the Dynamic Router {@link org.apache.camel.Endpoint} for the
given endpoint URI. This includes the
- * creation of a {@link DynamicRouterProcessor} that is registered
with the supplied
- * {@link DynamicRouterComponent}.
+ * Create the Dynamic Router {@link org.apache.camel.Endpoint} for the
given endpoint URI. When it is
+ * initialized, the endpoint gets the {@link DynamicRouterProcessor}
of its channel from the supplied
+ * {@link DynamicRouterComponent}, which creates it when the channel
has none.
*
* @param uri the endpoint URI
* @param component the Dynamic Router {@link
org.apache.camel.Component}
diff --git
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProcessor.java
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProcessor.java
index 51ab3afc5502..da5dc7e67059 100644
---
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProcessor.java
+++
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProcessor.java
@@ -27,6 +27,7 @@ import
org.apache.camel.component.dynamicrouter.filter.DynamicRouterFilterServic
import org.apache.camel.component.dynamicrouter.filter.PrioritizedFilter;
import org.apache.camel.processor.RecipientList;
import org.apache.camel.support.AsyncProcessorSupport;
+import org.apache.camel.support.service.ServiceHelper;
import static
org.apache.camel.component.dynamicrouter.routing.DynamicRouterConstants.MODE_FIRST_MATCH;
import static
org.apache.camel.component.dynamicrouter.routing.DynamicRouterConstants.RECIPIENT_LIST_HEADER;
@@ -136,6 +137,23 @@ public class DynamicRouterProcessor extends
AsyncProcessorSupport {
return recipientList.process(exchange, callback);
}
+ /**
+ * Stops the recipient list of this processor (which is started when the
processor is created), when the processor
+ * is replaced by the processor of an endpoint with other options. Its
thread pool is not shut down, so that the
+ * exchanges that are still in flight in it are not interrupted or dropped.
+ */
+ void stopRecipientList() {
+ ServiceHelper.stopService(recipientList);
+ }
+
+ /**
+ * The recipient list of this processor, which the component stops and
shuts down (with the thread pool it created)
+ * when the component shuts down.
+ */
+ RecipientList getRecipientList() {
+ return recipientList;
+ }
+
/**
* Factory for creating {@link DynamicRouterProcessor} instances.
*/
diff --git
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducer.java
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducer.java
index c8e01d4e580b..f1cc3496bdad 100644
---
a/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducer.java
+++
b/components/camel-dynamic-router/src/main/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducer.java
@@ -63,8 +63,7 @@ public class DynamicRouterProducer extends
DefaultAsyncProducer {
*/
@Override
public void process(final Exchange exchange) throws Exception {
- component.getRoutingProcessor(configuration.getChannel())
- .process(exchange);
+ routingProcessor().process(exchange);
}
/**
@@ -76,7 +75,14 @@ public class DynamicRouterProducer extends
DefaultAsyncProducer {
*/
@Override
public boolean process(final Exchange exchange, final AsyncCallback
callback) {
- DynamicRouterProcessor routingProcessor =
component.getRoutingProcessor(configuration.getChannel());
+ DynamicRouterProcessor routingProcessor;
+ try {
+ routingProcessor = routingProcessor();
+ } catch (IllegalStateException e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
if (configuration.isSynchronous()) {
try {
routingProcessor.process(exchange);
@@ -91,6 +97,21 @@ public class DynamicRouterProducer extends
DefaultAsyncProducer {
}
}
+ /**
+ * The processor of the channel, which the component keeps for the
endpoints of the channel.
+ *
+ * @throws IllegalStateException if the channel has no processor, such as
after the component has been shut down
+ */
+ private DynamicRouterProcessor routingProcessor() {
+ DynamicRouterProcessor routingProcessor =
component.getRoutingProcessor(configuration.getChannel());
+ if (routingProcessor == null) {
+ throw new IllegalStateException(
+ "The Dynamic Router channel '" + configuration.getChannel()
+ + "' has no processor");
+ }
+ return routingProcessor;
+ }
+
/**
* Create a {@link DynamicRouterProducer} instance.
*/
diff --git
a/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponentTest.java
b/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponentTest.java
index d4f6438ba74b..5c463eacf0e6 100644
---
a/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponentTest.java
+++
b/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterComponentTest.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.dynamicrouter.routing;
import java.util.Collections;
+import java.util.List;
import java.util.function.BiFunction;
import java.util.function.Supplier;
@@ -41,10 +42,13 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.mockito.Mock;
+import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class DynamicRouterComponentTest {
@@ -151,16 +155,51 @@ class DynamicRouterComponentTest {
}
@Test
- void testAddRoutingProcessor() {
- component.addRoutingProcessor(DYNAMIC_ROUTER_CHANNEL, processor);
+ void testGetOrCreateRoutingProcessor() throws Exception {
+ DynamicRouterConfiguration config = new DynamicRouterConfiguration();
+ config.setChannel(DYNAMIC_ROUTER_CHANNEL);
+ DynamicRouterEndpoint first =
Mockito.mock(DynamicRouterEndpoint.class);
+ DynamicRouterEndpoint sameUri =
Mockito.mock(DynamicRouterEndpoint.class);
+ DynamicRouterEndpoint otherUri =
Mockito.mock(DynamicRouterEndpoint.class);
+ for (DynamicRouterEndpoint e : List.of(first, sameUri, otherUri)) {
+ when(e.getConfiguration()).thenReturn(config);
+ }
+ when(first.getEndpointUri()).thenReturn("dynamic-router://test");
+ when(sameUri.getEndpointUri()).thenReturn("dynamic-router://test");
+
when(otherUri.getEndpointUri()).thenReturn("dynamic-router://test?recipientMode=allMatch");
+ DynamicRouterProcessor other =
Mockito.mock(DynamicRouterProcessor.class);
+
+ assertEquals(processor, component.getOrCreateRoutingProcessor(first,
() -> processor));
+ // an endpoint with the same uri uses the same processor
+ assertEquals(processor, component.getOrCreateRoutingProcessor(sameUri,
() -> other));
+ // an endpoint with another uri fails while the endpoint that uses the
processor is not stopped
+ assertThrows(IllegalArgumentException.class, () ->
component.getOrCreateRoutingProcessor(otherUri, () -> other));
assertEquals(processor,
component.getRoutingProcessor(DYNAMIC_ROUTER_CHANNEL));
- }
-
- @Test
- void testAddRoutingProcessorWithSecondProcessorForChannelError() {
- component.addRoutingProcessor(DYNAMIC_ROUTER_CHANNEL, processor);
+ Mockito.verify(processor, Mockito.never()).stopRecipientList();
+ // and replaces the processor once that endpoint is stopped; the
channel keeps its processor when the new one
+ // cannot be created
+ when(sameUri.isProcessorReleased()).thenReturn(true);
+ assertThrows(IllegalStateException.class, () ->
component.getOrCreateRoutingProcessor(otherUri, () -> {
+ throw new IllegalStateException("cannot create");
+ }));
assertEquals(processor,
component.getRoutingProcessor(DYNAMIC_ROUTER_CHANNEL));
- assertThrows(IllegalArgumentException.class, () ->
component.addRoutingProcessor(DYNAMIC_ROUTER_CHANNEL, processor));
+ Mockito.verify(processor, Mockito.never()).stopRecipientList();
+ assertEquals(other, component.getOrCreateRoutingProcessor(otherUri, ()
-> other));
+ assertEquals(other,
component.getRoutingProcessor(DYNAMIC_ROUTER_CHANNEL));
+ Mockito.verify(processor).stopRecipientList();
+
+ // the component shuts down the recipient lists of the processor of
the channel and of the replaced processor,
+ // all of them when one fails, and rethrows that failure
+ RecipientList replacedList = Mockito.mock(RecipientList.class);
+ RecipientList currentList = Mockito.mock(RecipientList.class);
+ when(processor.getRecipientList()).thenReturn(replacedList);
+ when(other.getRecipientList()).thenReturn(currentList);
+ Mockito.doThrow(new IllegalStateException("cannot
stop")).when(currentList).stop();
+ assertThrows(IllegalStateException.class, () ->
component.doShutdown());
+ Mockito.verify(replacedList).stop();
+ Mockito.verify(replacedList).shutdown();
+ Mockito.verify(currentList).stop();
+ assertNull(component.getRoutingProcessor(DYNAMIC_ROUTER_CHANNEL));
}
@Test
diff --git
a/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducerTest.java
b/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducerTest.java
index 54c13e79fd3c..f6155ac3c323 100644
---
a/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducerTest.java
+++
b/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterProducerTest.java
@@ -33,11 +33,13 @@ import org.mockito.junit.jupiter.MockitoExtension;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -106,6 +108,24 @@ class DynamicRouterProducerTest {
assertTrue(result);
}
+ @Test
+ void testProcessSynchronousWithoutProcessor() {
+ when(configuration.getChannel()).thenReturn("testChannel");
+ when(component.getRoutingProcessor("testChannel")).thenReturn(null);
+ IllegalStateException e = assertThrows(IllegalStateException.class, ()
-> producer.process(exchange));
+ assertTrue(e.getMessage().contains("'testChannel'"), e.getMessage());
+ }
+
+ @Test
+ void testProcessAsynchronousWithoutProcessor() {
+ when(configuration.getChannel()).thenReturn("testChannel");
+ when(component.getRoutingProcessor("testChannel")).thenReturn(null);
+ boolean result = producer.process(exchange, asyncCallback);
+ assertTrue(result);
+ verify(exchange).setException(any(IllegalStateException.class));
+ verify(asyncCallback).done(true);
+ }
+
@Test
void testGetInstance() {
DynamicRouterProducer instance = new DynamicRouterProducerFactory()
diff --git
a/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterRemoveRouteTest.java
b/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterRemoveRouteTest.java
new file mode 100644
index 000000000000..505930d9cad3
--- /dev/null
+++
b/components/camel-dynamic-router/src/test/java/org/apache/camel/component/dynamicrouter/routing/DynamicRouterRemoveRouteTest.java
@@ -0,0 +1,275 @@
+/*
+ * 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.dynamicrouter.routing;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.annotations.RouteFixture;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The channel of a Dynamic Router endpoint can be used again after the
endpoint was stopped, such as when the route
+ * that used it is removed.
+ */
+public class DynamicRouterRemoveRouteTest {
+
+ @RegisterExtension
+ static CamelContextExtension contextExtension = new
DefaultCamelContextExtension();
+
+ static volatile CountDownLatch reached = new CountDownLatch(1);
+
+ static volatile CountDownLatch release = new CountDownLatch(1);
+
+ CamelContext context;
+
+ ProducerTemplate template;
+
+ MockEndpoint mock;
+
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:subscribe")
+ .routeId("subscribe")
+
.to("dynamic-router-control:subscribe?subscribeChannel=orders&subscriptionId=all"
+ +
"&destinationUri=mock:result&priority=1&predicate=true")
+
.to("dynamic-router-control:subscribe?subscribeChannel=sync&subscriptionId=all"
+ +
"&destinationUri=mock:result&priority=1&predicate=true")
+
.to("dynamic-router-control:subscribe?subscribeChannel=par&subscriptionId=all"
+ +
"&destinationUri=direct:slow&priority=1&predicate=true");
+ from("direct:slow")
+ .routeId("slow")
+ .process(e -> {
+ reached.countDown();
+ release.await(10, TimeUnit.SECONDS);
+ })
+ .to("mock:slow");
+ from("direct:dynamic")
+ .routeId("dynamic")
+ .toD("dynamic-router:${header.channel}");
+ }
+ });
+ }
+
+ @BeforeEach
+ void setUp() throws Exception {
+ context = contextExtension.getContext();
+ template = contextExtension.getProducerTemplate();
+ mock = contextExtension.getMockEndpoint("mock:result");
+ mock.reset();
+ for (String id : new String[] { "router", "syncRouter",
"parallelRouter" }) {
+ if (context.getRoute(id) != null) {
+ context.getRouteController().stopRoute(id);
+ context.removeRoute(id);
+ }
+ }
+ template.sendBody("direct:subscribe", "");
+ context.addRoutes(routerRoutes());
+ }
+
+ private static RoutesBuilder routerRoutes() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").routeId("router").to("dynamic-router:orders");
+
from("direct:sync").routeId("syncRouter").to("dynamic-router:sync?synchronous=true");
+
from("direct:parallel").routeId("parallelRouter").to("dynamic-router:par?parallelProcessing=true");
+ }
+ };
+ }
+
+ private DynamicRouterComponent component() {
+ return context.getComponent("dynamic-router",
DynamicRouterComponent.class);
+ }
+
+ @Test
+ void addRouteAgainAfterRemove() throws Exception {
+ mock.expectedBodiesReceived("before", "after");
+
+ template.sendBody("direct:start", "before");
+ DynamicRouterProcessor processor =
component().getRoutingProcessor("orders");
+
+ context.getRouteController().stopRoute("router");
+ context.removeRoute("router");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").routeId("router").to("dynamic-router:orders");
+ }
+ });
+ template.sendBody("direct:start", "after");
+
+ mock.assertIsSatisfied();
+ // the endpoint of the route that is added again uses the processor of
the channel again
+ assertSame(processor, component().getRoutingProcessor("orders"));
+ }
+
+ @Test
+ void addRouteAgainWithOtherOptions() throws Exception {
+ mock.expectedBodiesReceived("before", "after");
+
+ template.sendBody("direct:start", "before");
+ DynamicRouterProcessor processor =
component().getRoutingProcessor("orders");
+
+ context.getRouteController().stopRoute("router");
+ context.removeRoute("router");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").routeId("router").to("dynamic-router:orders?recipientMode=allMatch");
+ }
+ });
+ template.sendBody("direct:start", "after");
+
+ mock.assertIsSatisfied();
+ // the endpoint with other options gets a new processor, and the
processor of the removed endpoint is stopped
+ assertNotSame(processor, component().getRoutingProcessor("orders"));
+ assertTrue(processor.getRecipientList().isStopped());
+ }
+
+ @Test
+ void addRouteAgainWithOtherOptionsKeepsThePoolOfTheReplacedProcessor()
throws Exception {
+ DynamicRouterProcessor processor =
component().getRoutingProcessor("par");
+
+ context.getRouteController().stopRoute("parallelRouter");
+ context.removeRoute("parallelRouter");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:parallel").routeId("parallelRouter")
+
.to("dynamic-router:par?parallelProcessing=true&recipientMode=allMatch");
+ }
+ });
+
+ // the recipient list of the replaced processor is stopped, but its
thread pool is not shut down, as another
+ // route could still have exchanges in flight in it
+ assertNotSame(processor, component().getRoutingProcessor("par"));
+ assertTrue(processor.getRecipientList().isStopped());
+
assertFalse(processor.getRecipientList().getExecutorService().isShutdown());
+ }
+
+ @Test
+ void toDAfterRemovedRoute() throws Exception {
+ mock.expectedBodiesReceived("before", "after");
+
+ // the toD uses the endpoint of the route
+ template.sendBodyAndHeader("direct:dynamic", "before", "channel",
"orders");
+ DynamicRouterProcessor processor =
component().getRoutingProcessor("orders");
+
+ context.getRouteController().stopRoute("router");
+ context.removeRoute("router");
+
+ // the toD creates the endpoint of the channel again
+ template.sendBodyAndHeader("direct:dynamic", "after", "channel",
"orders");
+
+ mock.assertIsSatisfied();
+ assertSame(processor, component().getRoutingProcessor("orders"));
+ }
+
+ @Test
+ void stopAndStartEndpoint() throws Exception {
+ mock.expectedBodiesReceived("after");
+ Endpoint endpoint = context.getEndpoint("dynamic-router:orders");
+ DynamicRouterProcessor processor =
component().getRoutingProcessor("orders");
+ assertNotNull(processor);
+
+ // a stopped endpoint is initialized again when it starts, and uses
the same processor
+ endpoint.stop();
+ endpoint.start();
+ assertSame(processor, component().getRoutingProcessor("orders"));
+
+ template.sendBody("direct:start", "after");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void secondEndpointForChannelKeepsTheProcessor() throws Exception {
+ mock.expectedBodiesReceived("after");
+ DynamicRouterProcessor processor =
component().getRoutingProcessor("orders");
+
+ // another endpoint for the channel fails as before while the endpoint
of the channel is in use
+ assertThrows(Exception.class, () ->
context.getEndpoint("dynamic-router:orders?recipientMode=allMatch"));
+
+ assertSame(processor, component().getRoutingProcessor("orders"));
+ assertFalse(processor.getRecipientList().isStopped());
+ template.sendBody("direct:start", "after");
+ mock.assertIsSatisfied();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "orders", "sync" })
+ void sendThroughRemovedEndpoint(String channel) throws Exception {
+ mock.expectedBodiesReceived("routed");
+ // a route that still sends to an endpoint that was removed (such as
an endpoint that a toD producer cache
+ // evicted while it was used) still routes with the processor of the
channel
+ context.removeEndpoint(context.getEndpoint(channel.equals("sync")
+ ? "dynamic-router:sync?synchronous=true" :
"dynamic-router:orders"));
+
+ Exchange out = template.send(channel.equals("sync") ? "direct:sync" :
"direct:start",
+ e -> e.getMessage().setBody("routed"));
+
+ assertNull(out.getException());
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void exchangeInFlightCompletesWhenEndpointIsRemoved() throws Exception {
+ reached = new CountDownLatch(1);
+ release = new CountDownLatch(1);
+ MockEndpoint slow = contextExtension.getMockEndpoint("mock:slow");
+ slow.reset();
+ slow.expectedBodiesReceived("in flight");
+
+ CompletableFuture<Exchange> future
+ = template.asyncSend("direct:parallel", e ->
e.getMessage().setBody("in flight"));
+ try {
+ assertTrue(reached.await(10, TimeUnit.SECONDS));
+
context.removeEndpoint(context.getEndpoint("dynamic-router:par?parallelProcessing=true"));
+ } finally {
+ release.countDown();
+ }
+
+ Exchange out = future.get(10, TimeUnit.SECONDS);
+ assertNull(out.getException());
+ slow.assertIsSatisfied();
+ }
+}
