This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26911-action-client-gone in repository https://gitbox.apache.org/repos/asf/camel.git
commit f3e69251a22ba76f2ef56d31c085d7de37740650 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Nov 13 15:41:30 2026 +0100 CAMEL-26911: camel-jbang - a command no longer waits for the answer of an integration that has stopped ActionClient waited until its timeout for the answer to an action, also when the integration had stopped (such as a camel cmd send waiting for the reply of a slow route when the integration is killed). It now watches the process of the integration, when it can see it, and ends with exit code 3 and "The integration with pid N stopped before it answered the action X" as soon as it is gone. An answer written just before it stopped is still read. executeText throws ActionClient.StoppedException, so the AI tools and RuntimeHelper say so instead of reporting a timeout. 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 | 50 +++++++++--- .../camel/dsl/jbang/core/common/RuntimeHelper.java | 2 + .../dsl/jbang/core/common/ActionClientTest.java | 92 +++++++++++++++++++++- 8 files changed, 145 insertions(+), 15 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 eb1094feafd4..770824db4733 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 @@ -2780,6 +2780,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 090900d1f096..abeccb3fcdc9 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 @@ -38,7 +38,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 { @@ -120,6 +120,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() { } @@ -191,6 +201,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", @@ -200,15 +214,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 { - return exchange(pid, versionOf(pid), request, timeout, true, text -> text).reply(); + Exchange<String> exchange = exchange(pid, versionOf(pid), request, timeout, true, text -> text); + if (exchange.stopped()) { + throw new StoppedException(pid, request.getString("action")); + } + return exchange.reply(); } - // 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( @@ -230,6 +254,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 @@ -243,16 +270,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 80740006e994..0b4af53c4c17 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,89 @@ 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); + } finally { + integration.destroyForcibly(); + } + } + @Test void olderIntegrationTakesAnActionWithoutConfirmingIt() throws Exception { // before 4.24 the connector deletes the request without answering an action without a result @@ -213,7 +299,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); @@ -226,9 +312,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(); } }
