davsclaus commented on code in PR #27430:
URL: https://github.com/apache/camel/pull/27430#discussion_r4210389026
##########
core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java:
##########
@@ -122,4 +125,273 @@ public void configure() {
assertMockEndpointsSatisfied();
}
+ @Test
+ public void testLocalConfigurationWithIndirectConsumer() throws Exception {
+ // CAMEL-25135: onCompletion from a named routeConfiguration must fire
even when
+ // the opted-in route is not itself the consumer route (e.g. REST DSL
→ direct:).
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer route — does NOT carry the routeConfigurationId
+ from("direct:consumer")
+ .to("direct:processor");
+
+ // processing route — opts in to the named configuration
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBothRoutesOptIn() throws Exception {
+ // CAMEL-25135: when both the consumer route and the called direct:
route
+ // opt in to the same named configuration, onCompletion must fire only
once
+ // per exchange, not once per opted-in route.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // both routes opt in to the same named configuration
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor");
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationSubRouteCalledTwice() throws Exception {
+ // CAMEL-25135: when the opted-in processor route is called twice in
the same
+ // exchange, onCompletion must fire only once per exchange, not once
per visit.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:consumer")
+ .to("direct:processor")
+ .to("direct:processor"); // called twice
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedMessageCount(2); // called twice
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerWithIndirectConsumer()
throws Exception {
+ // CAMEL-25135: BeforeConsumer mode with a named routeConfiguration
must fire
+ // when the opted-in route is not itself the consumer route.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer route — does NOT carry the routeConfigurationId
+ from("direct:consumer")
+ .to("direct:processor");
+
+ // processing route — opts in to the named configuration
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ // BeforeConsumer onCompletion runs before the consumer sends the
reply,
+ // so the header should be visible on the reply exchange
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerBothRoutesOptIn() throws
Exception {
+ // CAMEL-25135: BeforeConsumer + both routes opt in → fire only once.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor");
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
+ @Test
+ public void testGlobalStarConfiguration() throws Exception {
+ // routeConfiguration("*") is treated as global: fires once per
exchange for all routes
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+ routeConfiguration("*").onCompletion().to("mock:global");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .to("mock:result");
+
+ from("direct:start2")
+ .to("mock:result2");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:global").expectedMessageCount(2);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:result2").expectedBodiesReceived("Bye World");
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start2", "Bye World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testOnCompleteOnlyAndOnFailureOnlyInSameNamedConfig() throws
Exception {
+ // One named config with onCompleteOnly + onFailureOnly: they must not
dedup-block each other.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+ RouteConfigurationDefinition rc =
routeConfiguration("myconfig");
+ rc.onCompletion().onCompleteOnly().to("mock:complete");
+ rc.onCompletion().onFailureOnly().to("mock:failure");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeConfigurationId("myconfig")
+ .to("mock:result");
+
+ from("direct:fail").routeConfigurationId("myconfig")
+ .throwException(new IllegalArgumentException("Boom"));
+ }
+ });
+ context.start();
+
+ // success case
+ getMockEndpoint("mock:complete").expectedMessageCount(1);
+ getMockEndpoint("mock:failure").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:start", "Hello World");
+ assertMockEndpointsSatisfied();
+
+ // failure case
+ resetMocks();
+ getMockEndpoint("mock:complete").expectedMessageCount(0);
+ getMockEndpoint("mock:failure").expectedMessageCount(1);
+
+ try {
+ template.sendBody("direct:fail", "Boom");
+ } catch (Exception e) {
+ // expected
+ }
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerFiringCount() throws
Exception {
+ // BeforeConsumer mode, both routes opt in. onCompletion should fire
only once.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("count", constant(1));
Review Comment:
`constant(1)` gives the same result whether the onCompletion runs once or
twice. An `AtomicInteger` (or `.process(e -> counter.incrementAndGet())`)
asserted to be 1 would pin the count. It would also be good to cover the
deferral: e.g.
`from("direct:consumer")...to("direct:processor").setHeader("status",
constant("done"))` with an onCompletion
`onWhen(header("status").isEqualTo("done"))` that must fire.
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java:
##########
@@ -182,13 +197,28 @@ public void setStepId(String stepId) {
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
if (processor != null) {
+ // For named route configurations, register the route that owns
this processor
+ // so that BeforeConsumer can determine whether to defer to the
consumer route.
+ if (configurationId != null) {
+ @SuppressWarnings("unchecked")
+ Map<String, Set<String>> registeredRoutes
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_CONFIG_REGISTERED_ROUTES,
Map.class);
+ if (registeredRoutes == null) {
+ registeredRoutes = new HashMap<>();
+
exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_CONFIG_REGISTERED_ROUTES,
registeredRoutes);
+ }
+ registeredRoutes.computeIfAbsent(configurationId, k -> new
HashSet<>()).add(getRouteId());
+ }
Review Comment:
The registered-routes map is only read by the BeforeConsumer deferral, but
it is allocated and filled for every exchange with a named configuration,
including the default AfterConsumer mode. Guarding it with `!afterConsumer`
would keep the AfterConsumer hot path free of the extra `HashMap`/`HashSet` per
exchange.
Question: these maps and sets live in exchange properties, which
`Exchange.copy()` shares by reference, so with a parallel split/multicast into
an opted-in sub-route, could several threads mutate the same `HashMap` here?
`ON_COMPLETION_ROUTE_IDS` on main has the same shape, so this may be
acceptable, but it is worth a thought.
##########
core/camel-api/src/main/java/org/apache/camel/Exchange.java:
##########
@@ -255,6 +255,17 @@ public interface Exchange extends VariableAware {
description = "Flag to mark that this exchange is currently
being executed as onCompletion", javaType = "boolean")
String ON_COMPLETION = "CamelOnCompletion";
String ON_COMPLETION_ROUTE_IDS = "CamelOnCompletionRouteIds";
+ /**
+ * Exchange property key for tracking which named route configuration IDs
have already fired their onCompletion
+ * handler for this exchange. Used to ensure a named route configuration's
onCompletion fires at most once per
+ * exchange, even when multiple opted-in routes are traversed in the same
exchange.
+ */
+ String ON_COMPLETION_FIRED_CONFIG_IDS = "CamelOnCompletionFiredConfigIds";
Review Comment:
After this commit the set holds `OnCompletionProcessor` ids (one per
`OnCompletionDefinition`), not route configuration ids. Since the constant is
new in this PR, renaming it (e.g. `ON_COMPLETION_FIRED_IDS`) and updating the
Javadoc would keep it accurate.
##########
core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java:
##########
@@ -122,4 +125,273 @@ public void configure() {
assertMockEndpointsSatisfied();
}
+ @Test
+ public void testLocalConfigurationWithIndirectConsumer() throws Exception {
+ // CAMEL-25135: onCompletion from a named routeConfiguration must fire
even when
+ // the opted-in route is not itself the consumer route (e.g. REST DSL
→ direct:).
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer route — does NOT carry the routeConfigurationId
+ from("direct:consumer")
+ .to("direct:processor");
+
+ // processing route — opts in to the named configuration
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBothRoutesOptIn() throws Exception {
+ // CAMEL-25135: when both the consumer route and the called direct:
route
+ // opt in to the same named configuration, onCompletion must fire only
once
+ // per exchange, not once per opted-in route.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // both routes opt in to the same named configuration
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor");
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationSubRouteCalledTwice() throws Exception {
+ // CAMEL-25135: when the opted-in processor route is called twice in
the same
+ // exchange, onCompletion must fire only once per exchange, not once
per visit.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:consumer")
+ .to("direct:processor")
+ .to("direct:processor"); // called twice
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedMessageCount(2); // called twice
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerWithIndirectConsumer()
throws Exception {
+ // CAMEL-25135: BeforeConsumer mode with a named routeConfiguration
must fire
+ // when the opted-in route is not itself the consumer route.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer route — does NOT carry the routeConfigurationId
+ from("direct:consumer")
+ .to("direct:processor");
+
+ // processing route — opts in to the named configuration
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ // BeforeConsumer onCompletion runs before the consumer sends the
reply,
+ // so the header should be visible on the reply exchange
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerBothRoutesOptIn() throws
Exception {
+ // CAMEL-25135: BeforeConsumer + both routes opt in → fire only once.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor");
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
+ @Test
+ public void testGlobalStarConfiguration() throws Exception {
+ // routeConfiguration("*") is treated as global: fires once per
exchange for all routes
Review Comment:
Each exchange visits a single route here, so this passes whether `*` is
treated as global or as route-scoped. The case `*` must handle is one exchange
that goes `direct:start` → `direct:sub` and fires `mock:global` exactly once.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
