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 6d44dc033d8a CAMEL-26823: camel-core - the route topology says whether
a route waits for the answer of the route it sends to (#29599)
6d44dc033d8a is described below
commit 6d44dc033d8a27668117ffb76b65d95955e651bb
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Nov 11 11:37:20 2026 +0100
CAMEL-26823: camel-core - the route topology says whether a route waits for
the answer of the route it sends to (#29599)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/dev-consoles-openapi.json | 9 +-
.../camel/catalog/dev-consoles/route-topology.json | 9 +-
.../org/apache/camel/spi/RouteTopologyDumper.java | 16 ++-
.../apache/camel/dev-console/route-topology.json | 9 +-
.../impl/console/RouteTopologyDevConsole.java | 14 ++-
.../impl/console/RouteTopologyDevConsoleTest.java | 26 ++++-
.../camel/impl/DefaultDumpRoutesStrategy.java | 1 +
.../camel/impl/DefaultRouteTopologyDumper.java | 92 +++++++++++++++--
.../DefaultDumpRoutesStrategyTopologyJsonTest.java | 2 +
.../impl/DefaultRouteTopologyDumperReplyTest.java | 111 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +++
11 files changed, 285 insertions(+), 16 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
index d87ddb449973..14a36d69d974 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
@@ -7394,8 +7394,15 @@
"handling": {
"type": "string",
"description": "For an error path, what happens
to the failure: handled (the message ends there), continued (the route goes on
after it) or notHandled (the failure goes on to the caller); absent when a
predicate decides it"
+
},
+
"reply": {
+
"type": "boolean",
+
"description": "Whether the source route waits
for the answer of the target route: a call (direct) or a request-reply over a
broker, rather than a hand-off"
}
-
}
+
},
+
"required": [
+
"reply"
+
]
},
"description": "The connections between nodes (only present when a
route topology dumper is available)"
},
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/route-topology.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/route-topology.json
index 842512d8c341..976200187dbe 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/route-topology.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/route-topology.json
@@ -146,8 +146,15 @@
"handling": {
"type": "string",
"description": "For an error path, what happens to the failure:
handled (the message ends there), continued (the route goes on after it) or
notHandled (the failure goes on to the caller); absent when a predicate decides
it"
+ },
+ "reply": {
+ "type": "boolean",
+ "description": "Whether the source route waits for the answer of
the target route: a call (direct) or a request-reply over a broker, rather than
a hand-off"
}
- }
+ },
+ "required": [
+ "reply"
+ ]
},
"description": "The connections between nodes (only present when a
route topology dumper is available)"
},
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/RouteTopologyDumper.java
b/core/camel-api/src/main/java/org/apache/camel/spi/RouteTopologyDumper.java
index eb4a37340000..6ef8b42e6ed2 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/RouteTopologyDumper.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/RouteTopologyDumper.java
@@ -107,16 +107,28 @@ public interface RouteTopologyDumper {
* @param handling for an error path, what happens to the failure:
{@link #HANDLING_HANDLED},
* {@link #HANDLING_CONTINUED} or {@link
#HANDLING_NOT_HANDLED}; null for a call, and when a
* predicate decides it at runtime (since 4.23)
+ * @param reply whether the source route waits for the answer of
the target route: the send is InOut (the
+ * exchange pattern of the endpoint uri, else of the
send, else of the message the route
+ * consumed), or the component always waits, as
direct does: a request-reply over a broker
+ * rather than a hand-off (since 4.24)
* @since 4.21
*/
record TopologyEdge(String fromRouteId, String toRouteId, String endpoint,
String connectionType, String kind,
- String via, String handling) {
+ String via, String handling, boolean reply) {
/**
* An edge of kind {@link #EDGE_CALL}.
*/
public TopologyEdge(String fromRouteId, String toRouteId, String
endpoint, String connectionType) {
- this(fromRouteId, toRouteId, endpoint, connectionType, EDGE_CALL,
null, null);
+ this(fromRouteId, toRouteId, endpoint, connectionType, EDGE_CALL,
null, null, false);
+ }
+
+ /**
+ * An edge without the answer of its target known.
+ */
+ public TopologyEdge(String fromRouteId, String toRouteId, String
endpoint, String connectionType, String kind,
+ String via, String handling) {
+ this(fromRouteId, toRouteId, endpoint, connectionType, kind, via,
handling, false);
}
/**
diff --git
a/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/route-topology.json
b/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/route-topology.json
index 842512d8c341..976200187dbe 100644
---
a/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/route-topology.json
+++
b/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/route-topology.json
@@ -146,8 +146,15 @@
"handling": {
"type": "string",
"description": "For an error path, what happens to the failure:
handled (the message ends there), continued (the route goes on after it) or
notHandled (the failure goes on to the caller); absent when a predicate decides
it"
+ },
+ "reply": {
+ "type": "boolean",
+ "description": "Whether the source route waits for the answer of
the target route: a call (direct) or a request-reply over a broker, rather than
a hand-off"
}
- }
+ },
+ "required": [
+ "reply"
+ ]
},
"description": "The connections between nodes (only present when a
route topology dumper is available)"
},
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/RouteTopologyDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/RouteTopologyDevConsole.java
index 6d57c50284cb..b179ba34413f 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/RouteTopologyDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/RouteTopologyDevConsole.java
@@ -66,7 +66,9 @@ public class RouteTopologyDevConsole extends
AbstractDevConsole {
@Metadata(description = "For an error path, what sends:
errorHandler or onException") String via,
@Metadata(description = "For an error path, what happens to the
failure: handled (the message ends there),"
+ " continued (the route goes on after it)
or notHandled (the failure goes on to the"
- + " caller); absent when a predicate
decides it") String handling) {
+ + " caller); absent when a predicate
decides it") String handling,
+ @Metadata(description = "Whether the source route waits for the
answer of the target route: a call (direct)"
+ + " or a request-reply over a broker,
rather than a hand-off") boolean reply) {
}
public record ExternalEndpointEntry(
@@ -134,8 +136,12 @@ public class RouteTopologyDevConsole extends
AbstractDevConsole {
edge.toRouteId(), edge.endpoint(),
edge.connectionType(), edge.via(),
edge.handling() != null ? ", " +
edge.handling() : ""));
} else {
- sb.append(String.format(" --> %s via %s [%s]%n",
- edge.toRouteId(), edge.endpoint(),
edge.connectionType()));
+ // a request-reply the component does not imply
(direct always waits)
+ boolean requestReply = edge.reply() &&
!edge.endpoint().startsWith("direct:")
+ && !edge.endpoint().startsWith("kamelet:");
+ sb.append(String.format(" --> %s via %s [%s%s]%n",
+ edge.toRouteId(), edge.endpoint(),
edge.connectionType(),
+ requestReply ? ", request-reply" : ""));
}
}
}
@@ -186,7 +192,7 @@ public class RouteTopologyDevConsole extends
AbstractDevConsole {
for (TopologyEdge edge : result.edges()) {
edges.add(new EdgeEntry(
edge.fromRouteId(), edge.toRouteId(), edge.endpoint(),
edge.connectionType(), edge.kind(),
- edge.via(), edge.handling()));
+ edge.via(), edge.handling(), edge.reply()));
}
List<ExternalEndpointEntry> externalEndpoints = null;
diff --git
a/core/camel-console/src/test/java/org/apache/camel/impl/console/RouteTopologyDevConsoleTest.java
b/core/camel-console/src/test/java/org/apache/camel/impl/console/RouteTopologyDevConsoleTest.java
index f97c87fdd974..3be9d984f420 100644
---
a/core/camel-console/src/test/java/org/apache/camel/impl/console/RouteTopologyDevConsoleTest.java
+++
b/core/camel-console/src/test/java/org/apache/camel/impl/console/RouteTopologyDevConsoleTest.java
@@ -54,6 +54,13 @@ class RouteTopologyDevConsoleTest extends ContextTestSupport
{
from("direct:isolated").routeId("isolated")
.to("mock:nowhere");
+
+ // request-reply over a queue: the asker waits for the answer
of the quoter
+ from("direct:ask").routeId("asker")
+ .to("seda:quote?exchangePattern=InOut");
+
+ from("seda:quote").routeId("quoter")
+ .to("mock:quoted");
}
};
}
@@ -75,7 +82,7 @@ class RouteTopologyDevConsoleTest extends ContextTestSupport {
JsonArray nodes = (JsonArray) json.get("nodes");
assertNotNull(nodes);
- assertEquals(5, nodes.size());
+ assertEquals(7, nodes.size());
JsonArray edges = (JsonArray) json.get("edges");
assertNotNull(edges);
@@ -110,6 +117,23 @@ class RouteTopologyDevConsoleTest extends
ContextTestSupport {
assertEquals("internal", edge.getString("connectionType"));
}
+ @Test
+ void testAnEdgeSaysWhetherTheRouteWaitsForTheAnswer() {
+ DevConsole console =
PluginHelper.getDevConsoleResolver(context).resolveDevConsole("route-topology");
+ JsonObject json = (JsonObject) console.call(DevConsole.MediaType.JSON);
+ JsonArray edges = (JsonArray) json.get("edges");
+
+ // CAMEL-26823: a direct route runs before the sender goes on, a
request-reply waits for its answer
+ assertEquals(Boolean.TRUE, findEdge(edges, "trigger",
"processor").get("reply"));
+ assertEquals(Boolean.TRUE, findEdge(edges, "asker",
"quoter").get("reply"));
+ // a hand-off over a queue does not
+ assertEquals(Boolean.FALSE, findEdge(edges, "looper",
"looper").get("reply"));
+
+ String text = (String) console.call(DevConsole.MediaType.TEXT);
+ assertTrue(text.contains("--> quoter via seda:quote [internal,
request-reply]"), text);
+ assertTrue(text.contains("--> processor via direct:process
[internal]"), text);
+ }
+
@Test
void testChainedConnections() {
DevConsole console =
PluginHelper.getDevConsoleResolver(context).resolveDevConsole("route-topology");
diff --git
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultDumpRoutesStrategy.java
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultDumpRoutesStrategy.java
index 460928cd7603..6ef6c96fe064 100644
---
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultDumpRoutesStrategy.java
+++
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultDumpRoutesStrategy.java
@@ -341,6 +341,7 @@ public class DefaultDumpRoutesStrategy extends
ServiceSupport implements DumpRou
if (edge.handling() != null) {
jo.put("handling", edge.handling());
}
+ jo.put("reply", edge.reply());
edgesArr.add(jo);
}
root.put("edges", edgesArr);
diff --git
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultRouteTopologyDumper.java
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultRouteTopologyDumper.java
index 444c295f8bac..2e808385bd4f 100644
---
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultRouteTopologyDumper.java
+++
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultRouteTopologyDumper.java
@@ -28,6 +28,8 @@ import java.util.TreeMap;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.ErrorHandlerFactory;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Route;
import org.apache.camel.model.EndpointRequiredDefinition;
import org.apache.camel.model.ExpressionSubElementDefinition;
import org.apache.camel.model.Model;
@@ -35,12 +37,16 @@ import org.apache.camel.model.OnExceptionDefinition;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.ProcessorDefinitionHelper;
import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.SendDefinition;
+import org.apache.camel.model.ToDynamicDefinition;
+import org.apache.camel.model.WireTapDefinition;
import org.apache.camel.model.errorhandler.DeadLetterChannelDefinition;
import org.apache.camel.model.errorhandler.NoErrorHandlerDefinition;
import org.apache.camel.model.language.ConstantExpression;
import org.apache.camel.spi.EndpointUriFactory;
import org.apache.camel.spi.RouteTopologyDumper;
import org.apache.camel.spi.annotations.JdkService;
+import org.apache.camel.support.EndpointHelper;
import org.apache.camel.util.URISupport;
@JdkService(RouteTopologyDumper.FACTORY)
@@ -48,6 +54,8 @@ public class DefaultRouteTopologyDumper implements
RouteTopologyDumper {
private static final Set<String> TRIGGER_SCHEMES = Set.of("timer",
"quartz", "cron", "scheduler");
private static final Set<String> INTERNAL_SCHEMES = Set.of("direct",
"seda", "vertx", "disruptor", "disruptor-vm");
+ // the schemes whose target runs before the route goes on, whatever the
exchange pattern
+ private static final Set<String> WAITING_SCHEMES = Set.of("direct",
"kamelet");
@Override
public TopologyResult dumpTopology(CamelContext context) {
@@ -88,14 +96,23 @@ public class DefaultRouteTopologyDumper implements
RouteTopologyDumper {
Collection<EndpointRequiredDefinition> sends
=
ProcessorDefinitionHelper.filterTypeInOutputs(List.of(output),
EndpointRequiredDefinition.class);
for (EndpointRequiredDefinition erd : sends) {
+ // an error path ends there: what it sends to is no answer
the route waits for
+ boolean reply = !onException && waitsForAnswer(context,
rd, (ProcessorDefinition<?>) erd,
+ erd.getEndpointUri());
addEdges(context, edges, inputUriToRouteIds,
rd.getRouteId(), erd.getEndpointUri(), kind, via,
- handling);
+ handling, reply);
+ }
+ // a toD or a wire tap to an endpoint the route names, not one
it computes
+ for (ToDynamicDefinition td : staticDynamicSends(output)) {
+ boolean reply = !onException && waitsForAnswer(context,
rd, td, td.getUri());
+ addEdges(context, edges, inputUriToRouteIds,
rd.getRouteId(), td.getUri(), kind, via, handling,
+ reply);
}
}
if (handlesFailures && eh instanceof DeadLetterChannelDefinition
dlc && dlc.getDeadLetterUri() != null) {
// a dead letter channel handles the failure: the message ends
at the dead letter uri
addEdges(context, edges, inputUriToRouteIds, rd.getRouteId(),
dlc.getDeadLetterUri(), EDGE_ERROR,
- VIA_ERROR_HANDLER, HANDLING_HANDLED);
+ VIA_ERROR_HANDLER, HANDLING_HANDLED, false);
}
}
@@ -107,18 +124,73 @@ public class DefaultRouteTopologyDumper implements
RouteTopologyDumper {
private static void addEdges(
CamelContext context, List<TopologyEdge> edges, Map<String,
List<String>> inputUriToRouteIds,
- String fromRouteId, String uri, String kind, String via, String
handling) {
+ String fromRouteId, String uri, String kind, String via, String
handling, boolean reply) {
String outputUri = normalizeUri(context, uri);
List<String> targetRouteIds = inputUriToRouteIds.get(outputUri);
if (targetRouteIds != null) {
String scheme = extractScheme(outputUri);
String connType = INTERNAL_SCHEMES.contains(scheme) ? "internal" :
"external";
for (String targetId : targetRouteIds) {
- edges.add(new TopologyEdge(fromRouteId, targetId, outputUri,
connType, kind, via, handling));
+ edges.add(new TopologyEdge(fromRouteId, targetId, outputUri,
connType, kind, via, handling, reply));
}
}
}
+ /**
+ * Whether the route waits for the answer of what it sends to, as the send
processor decides the exchange pattern:
+ * the exchangePattern of the endpoint uri, else the pattern of the send
(to with pattern InOut, a wire tap is
+ * InOnly), else the pattern of the message the route consumed (InOut from
platform-http, InOnly from a timer).
+ * Direct always runs the target before the route goes on.
+ */
+ static boolean waitsForAnswer(CamelContext context, RouteDefinition route,
ProcessorDefinition<?> send, String uri) {
+ if (uri == null || send instanceof WireTapDefinition) {
+ // a wire tap sends a copy on another thread: the route goes on
without its answer
+ return false;
+ }
+ if (WAITING_SCHEMES.contains(extractScheme(uri))) {
+ return true;
+ }
+ ExchangePattern pattern = null;
+ try {
+ pattern =
EndpointHelper.resolveExchangePatternFromUrl(context.resolvePropertyPlaceholders(uri));
+ } catch (Exception e) {
+ // a uri that cannot be resolved says nothing
+ }
+ if (pattern == null) {
+ String sendPattern = send instanceof SendDefinition<?> sd ?
sd.getPattern()
+ : send instanceof ToDynamicDefinition td ? td.getPattern()
: null;
+ if (sendPattern != null) {
+ try {
+ pattern =
context.getTypeConverter().tryConvertTo(ExchangePattern.class,
+ context.resolvePropertyPlaceholders(sendPattern));
+ } catch (Exception e) {
+ // a pattern that cannot be resolved says nothing
+ }
+ }
+ }
+ if (pattern == null) {
+ Route running = route.getRouteId() != null ?
context.getRoute(route.getRouteId()) : null;
+ pattern = running != null && running.getEndpoint() != null ?
running.getEndpoint().getExchangePattern() : null;
+ }
+ return pattern == ExchangePattern.InOut;
+ }
+
+ /**
+ * The toD and wire tap sends below a processor whose uri has no
expression: the route names the endpoint (a wire
+ * tap to seda:audit), as a to does.
+ */
+ private static List<ToDynamicDefinition>
staticDynamicSends(ProcessorDefinition<?> output) {
+ List<ToDynamicDefinition> answer = new ArrayList<>();
+ for (ToDynamicDefinition td :
ProcessorDefinitionHelper.filterTypeInOutputs(List.of(output),
+ ToDynamicDefinition.class)) {
+ String uri = td.getUri();
+ if (uri != null && !uri.contains("${") &&
!uri.contains("$simple{")) {
+ answer.add(td);
+ }
+ }
+ return answer;
+ }
+
/**
* What an onException clause does with the failure: continued, handled or
not handled (a side trip, the failure
* goes on to the caller); null when a predicate decides it at runtime.
@@ -172,6 +244,9 @@ public class DefaultRouteTopologyDumper implements
RouteTopologyDumper {
for (EndpointRequiredDefinition erd : outputs) {
allOutputUris.add(URISupport.stripQuery(erd.getEndpointUri()));
}
+ for (ProcessorDefinition<?> output : rd.getOutputs()) {
+ staticDynamicSends(output).forEach(td ->
allOutputUris.add(URISupport.stripQuery(td.getUri())));
+ }
}
List<TopologyExternalEndpoint> externalEndpoints = new ArrayList<>();
@@ -193,9 +268,14 @@ public class DefaultRouteTopologyDumper implements
RouteTopologyDumper {
= ProcessorDefinitionHelper.filterTypeInOutputs(
rd.getOutputs(), EndpointRequiredDefinition.class);
+ List<String> outputUris = new ArrayList<>();
+ outputs.forEach(erd -> outputUris.add(erd.getEndpointUri()));
+ for (ProcessorDefinition<?> output : rd.getOutputs()) {
+ staticDynamicSends(output).forEach(td ->
outputUris.add(td.getUri()));
+ }
int outIdx = 0;
- for (EndpointRequiredDefinition erd : outputs) {
- String outputUri = URISupport.stripQuery(erd.getEndpointUri());
+ for (String uri : outputUris) {
+ String outputUri = URISupport.stripQuery(uri);
String outputScheme = extractScheme(outputUri);
if (isRemoteScheme(outputScheme, schemeRemoteMap) &&
!inputUriToRouteIds.containsKey(outputUri)) {
String dedupeKey = routeId + "|" + outputUri;
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultDumpRoutesStrategyTopologyJsonTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultDumpRoutesStrategyTopologyJsonTest.java
index a60ceccade36..7e08639cbe80 100644
---
a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultDumpRoutesStrategyTopologyJsonTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultDumpRoutesStrategyTopologyJsonTest.java
@@ -80,6 +80,8 @@ public class DefaultDumpRoutesStrategyTopologyJsonTest
extends ContextTestSuppor
assertThat(edge.getString("toRouteId")).isEqualTo("consumer");
assertThat(edge.getString("endpoint")).isEqualTo("direct:linked");
assertThat(edge.getString("connectionType")).isEqualTo("internal");
+ // CAMEL-26823: the producer waits for the consumer, as direct runs it
before the producer goes on
+ assertThat(edge.get("reply")).isEqualTo(Boolean.TRUE);
}
@Test
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultRouteTopologyDumperReplyTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultRouteTopologyDumperReplyTest.java
new file mode 100644
index 000000000000..238408ba2c2d
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultRouteTopologyDumperReplyTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.impl;
+
+import java.util.List;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.ToDefinition;
+import org.apache.camel.spi.RouteTopologyDumper;
+import org.apache.camel.spi.RouteTopologyDumper.TopologyEdge;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * CAMEL-26823: an edge says whether the route waits for the answer of the
route it sends to, as the send processor
+ * decides the exchange pattern: the exchangePattern of the endpoint uri, else
the pattern of the send, else the pattern
+ * of the message the route consumed; a direct route always runs before the
sender goes on. The shape of
+ * connect-service/jms-request-reply, with seda in the place of the JMS broker.
+ */
+class DefaultRouteTopologyDumperReplyTest extends ContextTestSupport {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("timer:orders?repeatCount=1").routeId("order-intake")
+ // a hand-off: the order goes on a queue
+ .to("seda:orders");
+
+ from("seda:orders").routeId("reserve-order")
+ .errorHandler(deadLetterChannel("seda:orders-dead"))
+ // request-reply: the endpoint asks for the answer
+ .to("seda:stock?exchangePattern=InOut")
+ // the send asks for it
+ .to(ExchangePattern.InOut, "seda:pricing")
+ // a call
+ .to("direct:audit")
+ // a copy, never an answer
+ .wireTap("seda:metrics")
+ // a hand-off to the subscribers
+ .to("seda:reserved");
+
+ // the request of a client: what it sends on is InOut too
+ from("seda:quote?exchangePattern=InOut").routeId("quote")
+ .to("seda:pricing");
+
+ from("seda:stock").routeId("stock-service").log("stock");
+ from("seda:pricing").routeId("pricing").log("pricing");
+ from("direct:audit").routeId("audit").log("audit");
+ from("seda:metrics").routeId("metrics").log("metrics");
+ from("seda:reserved").routeId("shipping").log("shipping");
+ from("seda:orders-dead").routeId("dead-orders").log("dead");
+ }
+ };
+ }
+
+ @Test
+ void anEdgeSaysWhetherTheRouteWaitsForTheAnswer() {
+ List<TopologyEdge> edges = new
DefaultRouteTopologyDumper().dumpTopology(context).edges();
+
+ assertThat(edges).extracting(e -> e.fromRouteId() + "->" +
e.toRouteId() + (e.reply() ? " waits" : ""))
+ .contains(
+ "order-intake->reserve-order",
+ "reserve-order->stock-service waits",
+ "reserve-order->pricing waits",
+ "reserve-order->audit waits",
+ "reserve-order->metrics",
+ "reserve-order->shipping",
+ "quote->pricing waits");
+ // the dead letter channel: the message ends there
+ assertThat(edges).filteredOn(e -> "dead-orders".equals(e.toRouteId()))
+ .isNotEmpty().allMatch(e -> e.isErrorPath() && !e.reply());
+ }
+
+ @Test
+ void aPatternThatCannotBeResolvedSaysNothing() {
+ // the topology is dumped from the model, so a send pattern with a
missing placeholder must not fail the dump
+ RouteDefinition route = new RouteDefinition();
+ route.routeId("not-started");
+ ToDefinition send = new ToDefinition("seda:pricing");
+ send.setPattern("{{missing.pattern}}");
+
+ assertThat(DefaultRouteTopologyDumper.waitsForAnswer(context, route,
send, "seda:pricing")).isFalse();
+ }
+
+ @Test
+ void anEdgeOfBeforeWaitsForNoAnswer() {
+ assertThat(new TopologyEdge("a", "b", "seda:b",
"internal").reply()).isFalse();
+ assertThat(new TopologyEdge("a", "b", "seda:b", "internal",
RouteTopologyDumper.EDGE_CALL, null, null).reply())
+ .isFalse();
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 58292cacc40c..be4266e014ba 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -256,6 +256,18 @@ in its model, so clearing them on the route saved little
memory at the cost of m
Applications that called `Route.clearRouteModel()` directly can remove the
call. Implementations of the `Route` interface
that override `clearRouteModel()` can remove the override.
+=== camel-core - the route topology says whether a route waits for the answer,
and has the edges of a wire tap
+
+Each edge of the route topology (the `route-topology` developer console, the
topology JSON of the dump routes strategy
+and `RouteTopologyDumper`) now says whether the route waits for the answer of
the route it sends to: `reply` is `true`
+for a send to `direct`, and for a send that is InOut, as a request-reply over
a JMS queue is
+(`exchangePattern=InOut`). The JSON of the console and the topology JSON write
`reply` on every edge, and the text
+output of the console marks a request-reply with `request-reply`. The
`RouteTopologyDumper.TopologyEdge` record has the
+new `reply` component; its seven-argument constructor is kept and creates an
edge without it.
+
+A `wireTap` or a `toD` to an endpoint the route names, rather than one it
computes with an expression, is now an edge
+too (a wire tap never waits for the answer), and an external endpoint when no
route consumes from it.
+
=== camel-management - LastExchangeCompletedExchangeId is updated at most once
per millisecond
The JMX attribute `LastExchangeCompletedExchangeId`, of the `CamelContext`,
route and processor MBeans, is now updated
