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 5ba1da5223b0 CAMEL-26911: camel-jbang - a command no longer waits for
the answer of an integration that has stopped (#29853)
5ba1da5223b0 is described below
commit 5ba1da5223b0f0f440b963fd35be4fe55775d166
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Nov 13 18:43:13 2026 +0100
CAMEL-26911: camel-jbang - a command no longer waits for the answer of an
integration that has stopped (#29853)
The Camel CLI waited for the answer of an integration to an action (camel
cmd send --reply, camel cmd receive, a route dump, the tools of camel ask and
the TUI, ...) until its timeout, also when the integration had stopped in the
meantime.
ActionClient now watches the process of the integration while it waits, and
ends as soon as it is gone, with exit code 3 as for an integration that is not
running. The process is checked before the answer is read, so an answer written
just before the integration stopped is still read. Only a process that is alive
when the action is sent is watched: an integration the CLI cannot see as a
process is waited for until the timeout, as before.
ActionClient.executeText throws ActionClient.StoppedException, so camel
ask, the TUI and RuntimeHelper.executeAction say that the integration stopped
instead of reporting a timeout. camel cmd send and camel transform message put
"Send timeout:" / "Transform timeout:" in front of the error only when it is a
timeout.
Docs: the exit codes page and the 4.24 upgrade guide.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 2 +
.../modules/ROOT/pages/camel-jbang-exit-codes.adoc | 3 +-
.../core/commands/action/CamelSendAction.java | 4 +-
.../commands/action/TransformMessageAction.java | 5 +-
.../dsl/jbang/core/commands/ai/ToolContext.java | 2 +
.../camel/dsl/jbang/core/common/ActionClient.java | 52 +++++++++---
.../camel/dsl/jbang/core/common/RuntimeHelper.java | 2 +
.../dsl/jbang/core/common/ActionClientTest.java | 95 +++++++++++++++++++++-
8 files changed, 148 insertions(+), 17 deletions(-)
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 9375415a3610..e5a6622f8a52 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
@@ -2829,6 +2829,8 @@ xref:camel-jbang-exit-codes.adoc[Exit Codes and Output]):
* A name or pid that matches no running integration exits with `3`, instead of
doing nothing and exiting with `0`.
Without a name (all integrations) and nothing running, the command exits
with `0` and prints
`No running Camel integrations` on stderr.
+* An integration that stops before it answers the action exits with `3` as
soon as it is gone, instead of waiting until
+ `--timeout` and exiting with `124`, such as a `camel cmd send` that waits
for the reply of a slow route.
* `camel eval expression` with a name or pid that matches no running
integration exits with `3`, instead of starting
a new Camel to evaluate in: only without a name, or with `--isolated`, a new
Camel evaluates the expression. An
evaluation that fails exits with `1`, and the `Using existing running Camel
integration` line is printed on stderr.
diff --git a/docs/user-manual/modules/ROOT/pages/camel-jbang-exit-codes.adoc
b/docs/user-manual/modules/ROOT/pages/camel-jbang-exit-codes.adoc
index 83b9584a4daa..6b204965dbde 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-jbang-exit-codes.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-exit-codes.adoc
@@ -29,7 +29,8 @@ The exit codes and streams on this page apply from Camel 4.24
onwards.
| `3`
| Not found
| An integration, route, component, service or other name given on the command
line is not running or does not exist,
- or a name matches more than one running integration.
+ or a name matches more than one running integration. Also when the
integration stops before it answers the command,
+ such as a `camel cmd send` that waits for the reply.
| `66`
| File not found
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
index 21a0fc9f7e95..9dd35d79cc17 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
@@ -306,7 +306,9 @@ public class CamelSendAction extends ActionBaseCommand {
JsonObject jo = result.reply();
if (jo == null) {
// no answer from the integration at all
- throw new CliException(result.exitCode(), "Send timeout: " +
result.error());
+ // an integration that stopped before it answered is not a timeout
+ String error = result.status() == ActionClient.Status.TIMEOUT ?
"Send timeout: " + result.error() : result.error();
+ throw new CliException(result.exitCode(), error);
}
printStatusLine(jo);
if (jo.containsKey("sent")) {
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/TransformMessageAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/TransformMessageAction.java
index 6a52f2f16261..aa9c9840f189 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/TransformMessageAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/TransformMessageAction.java
@@ -235,7 +235,10 @@ public class TransformMessageAction extends
ActionWatchCommand {
ActionResult result = ActionClient.execute(pid, root, timeout);
JsonObject jo = result.reply();
if (jo == null) {
- throw new CliException(result.exitCode(), "Transform timeout: " +
result.error());
+ // an integration that stopped before it answered is not a timeout
+ String error
+ = result.status() == ActionClient.Status.TIMEOUT ?
"Transform timeout: " + result.error() : result.error();
+ throw new CliException(result.exitCode(), error);
}
printStatusLine(jo);
String exchangeId = jo.getString("exchangeId");
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
index 7d56b3725e23..16d271524594 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
@@ -342,6 +342,8 @@ public class ToolContext {
String answer;
try {
answer = ActionClient.executeText(pid, request, ACTION_TIMEOUT);
+ } catch (ActionClient.StoppedException e) {
+ throw new ToolExecutionException(e.getMessage());
} catch (IOException e) {
throw new ToolExecutionException(
"Cannot send the action " + action + " to the integration
with pid " + pid + ": " + e.getMessage());
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ActionClient.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ActionClient.java
index ed751bc19efb..07c54b62a28b 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ActionClient.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ActionClient.java
@@ -39,7 +39,7 @@ import org.apache.camel.util.json.Jsoner;
* <tt>{pid}-output-{id}.json</tt>), so clients never take each other's
answers; an older integration only reads the
* shared <tt>{pid}-action.json</tt>. An integration on Camel 4.24 or later
answers every action, also one without a
* result, so no answer is a timeout; an older integration does not answer an
action without a result, which is then
- * sent but not confirmed.
+ * sent but not confirmed. An integration that stops before it answers is not
waited for until the timeout.
*/
public final class ActionClient {
@@ -121,6 +121,16 @@ public final class ActionClient {
}
}
+ /**
+ * The integration stopped before it answered the action.
+ */
+ public static final class StoppedException extends IOException {
+
+ public StoppedException(long pid, String action) {
+ super(stopped(pid, action));
+ }
+ }
+
private ActionClient() {
}
@@ -192,6 +202,10 @@ public final class ActionClient {
return new ActionResult(Status.UNCONFIRMED, null, null, null);
}
String action = request.getString("action");
+ if (exchange.stopped()) {
+ // exit code 3, as for an integration that is not running
+ return new ActionResult(Status.ERROR, null, stopped(pid, action),
"not-found");
+ }
return new ActionResult(
Status.TIMEOUT, null, "No answer from the integration with pid
" + pid + " to the action " + action
+ " within " + timeout + " millis",
@@ -201,20 +215,25 @@ public final class ActionClient {
/**
* Runs an action and returns the answer as the integration wrote it, for
tools that pass it on as text.
*
- * @return the answer, or <tt>null</tt> when there is none in
time
- * @throws IOException when the action cannot be written
+ * @return the answer, or <tt>null</tt> when there is
none in time
+ * @throws StoppedException when the integration stopped before it answered
+ * @throws IOException when the action cannot be written
*/
public static String executeText(long pid, JsonObject request, long
timeout) throws IOException {
// the answer once it is all there, as an integration of an earlier
Camel version writes it in place
// (CAMEL-26911): JSON once it parses, other text once it reads the
same at two polls
AtomicReference<String> previous = new AtomicReference<>();
- return exchange(pid, versionOf(pid), request, timeout, true, text -> {
+ Exchange<String> exchange = exchange(pid, versionOf(pid), request,
timeout, true, text -> {
if (isJson(text)) {
return text;
}
boolean json = text.stripLeading().startsWith("{") ||
text.stripLeading().startsWith("[");
return !json && text.equals(previous.getAndSet(text)) ? text :
null;
- }).reply();
+ });
+ if (exchange.stopped()) {
+ throw new StoppedException(pid, request.getString("action"));
+ }
+ return exchange.reply();
}
private static boolean isJson(String text) {
@@ -226,8 +245,13 @@ public final class ActionClient {
}
}
- // the answer to an action, as read, or null; and whether the integration
took the action
- private record Exchange<T>(T reply, boolean taken) {
+ private static String stopped(long pid, String action) {
+ return "The integration with pid " + pid + " stopped before it
answered the action " + action;
+ }
+
+ // the answer to an action, as read, or null; whether the integration took
the action; and whether it stopped before
+ // it answered
+ private record Exchange<T>(T reply, boolean taken, boolean stopped) {
}
private static <T> Exchange<T> exchange(
@@ -249,6 +273,9 @@ public final class ActionClient {
Files.deleteIfExists(outputFile);
}
Path temporary = dir.resolve(pid + "-" + UUID.randomUUID() + ".tmp");
+ // an integration that cannot be seen from here, such as one in a
container that shares the Camel directory, is
+ // waited for until the timeout
+ ProcessHandle process =
ProcessHandle.of(pid).filter(ProcessHandle::isAlive).orElse(null);
try {
Files.writeString(temporary, request.toJson());
// the integration must never read half an action
@@ -262,16 +289,21 @@ public final class ActionClient {
Thread.currentThread().interrupt();
break;
}
+ // before the answer is read, as an integration may answer
just before it stops
+ boolean alive = process == null || process.isAlive();
if (waitForAnswer) {
T reply = readReply(outputFile, reader);
if (reply != null) {
- return new Exchange<>(reply, true);
+ return new Exchange<>(reply, true, false);
}
} else if (!Files.exists(actionFile)) {
- return new Exchange<>(null, true);
+ return new Exchange<>(null, true, false);
+ }
+ if (!alive) {
+ return new Exchange<>(null, false, true);
}
}
- return new Exchange<>(null, false);
+ return new Exchange<>(null, false, false);
} finally {
// removing an action the integration has not taken yet cancels it
Files.deleteIfExists(temporary);
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
index f6411741f8bb..ca976164a133 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
@@ -167,6 +167,8 @@ public final class RuntimeHelper {
if (response != null) {
return response;
}
+ } catch (ActionClient.StoppedException e) {
+ return e.getMessage();
} catch (IOException e) {
// Preserve the existing text API's diagnostic when the request
cannot be exchanged.
}
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ActionClientTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ActionClientTest.java
index 2a3d7388f2bd..d08a7795f167 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ActionClientTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ActionClientTest.java
@@ -28,11 +28,14 @@ import java.util.function.Function;
import org.apache.camel.dsl.jbang.core.common.ActionClient.ActionResult;
import org.apache.camel.dsl.jbang.core.common.ActionClient.Status;
import org.apache.camel.dsl.jbang.core.common.ProcessFinder.CamelProcess;
+import org.apache.camel.util.StopWatch;
import org.apache.camel.util.json.JsonObject;
import org.apache.camel.util.json.Jsoner;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.io.TempDir;
import static org.assertj.core.api.Assertions.assertThat;
@@ -124,6 +127,90 @@ class ActionClientTest {
assertNoFilesLeft();
}
+ @Test
+ @DisabledOnOs(OS.WINDOWS)
+ void integrationThatStopsIsNotWaitedFor() throws Exception {
+ // a process of its own, as the client only watches a process it can
see
+ Process integration = new ProcessBuilder("sleep", "30").start();
+ try {
+ long pid = integration.pid();
+ executor.submit(() -> {
+ // stops while it has the action (such as a send that waits
for a reply), without answering it
+ await().atMost(5, TimeUnit.SECONDS).until(() ->
!requestFiles(CommandLineHelper.getCamelDir(), pid).isEmpty());
+ integration.destroy();
+ return null;
+ });
+ JsonObject request = new JsonObject();
+ request.put("action", "send");
+ StopWatch watch = new StopWatch();
+
+ ActionResult result = ActionClient.execute(pid, request, 30000);
+
+ assertThat(watch.taken()).isLessThan(10000);
+ assertThat(result.status()).isEqualTo(Status.ERROR);
+ assertThat(result.exitCode()).isEqualTo(CliExitCode.NOT_FOUND);
+ assertThat(result.error())
+ .isEqualTo("The integration with pid " + pid + " stopped
before it answered the action send");
+ assertNoFilesLeft();
+ } finally {
+ integration.destroyForcibly();
+ }
+ }
+
+ @Test
+ @DisabledOnOs(OS.WINDOWS)
+ void textOfAnIntegrationThatStopsIsAnError() throws Exception {
+ Process integration = new ProcessBuilder("sleep", "30").start();
+ try {
+ long pid = integration.pid();
+ executor.submit(() -> {
+ await().atMost(5, TimeUnit.SECONDS).until(() ->
!requestFiles(CommandLineHelper.getCamelDir(), pid).isEmpty());
+ integration.destroy();
+ return null;
+ });
+ JsonObject request = new JsonObject();
+ request.put("action", "route-dump");
+
+ // the tools of camel ask and the TUI say so, instead of that the
integration did not answer in time
+ assertThatThrownBy(() -> ActionClient.executeText(pid, request,
30000))
+ .isInstanceOf(ActionClient.StoppedException.class)
+ .hasMessage("The integration with pid " + pid + " stopped
before it answered the action route-dump");
+ assertNoFilesLeft();
+ } finally {
+ integration.destroyForcibly();
+ }
+ }
+
+ @Test
+ @DisabledOnOs(OS.WINDOWS)
+ void answerOfAnIntegrationThatStopsRightAfterIsRead() throws Exception {
+ Process integration = new ProcessBuilder("sleep", "30").start();
+ try {
+ long pid = integration.pid();
+ executor.submit(() -> {
+ Path dir = CommandLineHelper.getCamelDir();
+ Path request = await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> requestFiles(dir, pid), files ->
!files.isEmpty()).get(0);
+ String id = request.getFileName().toString().substring((pid +
"-action-").length()).replace(".json", "");
+ Files.writeString(dir.resolve(pid + "-output-" + id +
".json"), "{\"status\":\"success\",\"value\":42}");
+ integration.destroy();
+ integration.waitFor(5, TimeUnit.SECONDS);
+ Files.deleteIfExists(request);
+ return null;
+ });
+ JsonObject request = new JsonObject();
+ request.put("action", "jvm");
+
+ ActionResult result = ActionClient.execute(pid, request, 30000);
+
+ assertThat(result.status()).isEqualTo(Status.OK);
+ assertThat(result.reply().getInteger("value")).isEqualTo(42);
+ assertNoFilesLeft();
+ } finally {
+ integration.destroyForcibly();
+ }
+ }
+
@Test
void olderIntegrationTakesAnActionWithoutConfirmingIt() throws Exception {
// before 4.24 the connector deletes the request without answering an
action without a result
@@ -177,7 +264,7 @@ class ActionClientTest {
// an integration of an earlier Camel version writes the answer in
place: half of it is there first
executor.submit(() -> {
Path dir = CommandLineHelper.getCamelDir();
- Path request = await().atMost(5, TimeUnit.SECONDS).until(() ->
requestFiles(dir), files -> !files.isEmpty())
+ Path request = await().atMost(5, TimeUnit.SECONDS).until(() ->
requestFiles(dir, PID), files -> !files.isEmpty())
.get(0);
String id = request.getFileName().toString().substring((PID +
"-action-").length()).replace(".json", "");
Path output = dir.resolve(PID + "-output-" + id + ".json");
@@ -238,7 +325,7 @@ class ActionClientTest {
private Future<JsonObject> answerRequest(Function<JsonObject, JsonObject>
answer) {
return executor.submit(() -> {
Path dir = CommandLineHelper.getCamelDir();
- Path request = await().atMost(5, TimeUnit.SECONDS).until(() ->
requestFiles(dir), files -> !files.isEmpty())
+ Path request = await().atMost(5, TimeUnit.SECONDS).until(() ->
requestFiles(dir, PID), files -> !files.isEmpty())
.get(0);
JsonObject action = json(Files.readString(request));
JsonObject reply = answer.apply(action);
@@ -251,9 +338,9 @@ class ActionClientTest {
});
}
- private static List<Path> requestFiles(Path dir) throws Exception {
+ private static List<Path> requestFiles(Path dir, long pid) throws
Exception {
try (var files = Files.list(dir)) {
- return files.filter(f -> f.getFileName().toString().startsWith(PID
+ "-action-")).toList();
+ return files.filter(f -> f.getFileName().toString().startsWith(pid
+ "-action-")).toList();
}
}
