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 c9c4f3fd7e78 CAMEL-25447: camel-cli-connector - answer actions in
about 100 ms instead of once a second (#29727)
c9c4f3fd7e78 is described below
commit c9c4f3fd7e787f0571fb9562fa5ac7082a293403
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Nov 12 19:28:46 2026 +0100
CAMEL-25447: camel-cli-connector - answer actions in about 100 ms instead
of once a second (#29727)
The file transport of camel-cli-connector answers actions in about 100 ms
instead of up to a second (camel get / camel cmd, the dashboard, the MCP
server
and camel ask). A second task on the same single-thread executor looks for
action files every 100 ms, while the status is still written every second.
It
is only scheduled when the interval is above 100 ms. updateDelay and doStop
cancel and reschedule both tasks. Docs of the file transport updated.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/cli-connector.adoc | 3 ++
.../src/main/docs/cli-connector.adoc | 3 ++
.../camel/cli/connector/CliConnectorTransport.java | 3 +-
.../apache/camel/cli/connector/ConnectorTasks.java | 2 +-
.../cli/connector/FileCliConnectorTransport.java | 63 ++++++++++++++++++----
.../connector/FileCliConnectorTransportTest.java | 23 ++++++++
6 files changed, 85 insertions(+), 12 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cli-connector.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cli-connector.adoc
index 6e07c079f9e3..da55ff3487a6 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cli-connector.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cli-connector.adoc
@@ -62,6 +62,9 @@ route. When the action wrote a result before it failed, the
answer has it under
two routes that fail to stop, are joined with `;` in `error`. A result can
also report a failure of its own, such as a
message sent to a route that throws an exception: it is written as it is.
+The connector looks for action files about every 100 millis, and writes the
status to `~/.camel/<pid>-status.json`
+every second.
+
Write the action file atomically (write a temporary file, then move it), so
the connector never reads half an action.
An action file that cannot be read yet is retried for two seconds. Deleting
the action file before the answer is
written cancels the action.
diff --git a/dsl/camel-cli-connector/src/main/docs/cli-connector.adoc
b/dsl/camel-cli-connector/src/main/docs/cli-connector.adoc
index 6e07c079f9e3..da55ff3487a6 100644
--- a/dsl/camel-cli-connector/src/main/docs/cli-connector.adoc
+++ b/dsl/camel-cli-connector/src/main/docs/cli-connector.adoc
@@ -62,6 +62,9 @@ route. When the action wrote a result before it failed, the
answer has it under
two routes that fail to stop, are joined with `;` in `error`. A result can
also report a failure of its own, such as a
message sent to a route that throws an exception: it is written as it is.
+The connector looks for action files about every 100 millis, and writes the
status to `~/.camel/<pid>-status.json`
+every second.
+
Write the action file atomically (write a temporary file, then move it), so
the connector never reads half an action.
An action file that cannot be read yet is retried for two seconds. Deleting
the action file before the answer is
written cancels the action.
diff --git
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/CliConnectorTransport.java
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/CliConnectorTransport.java
index e207990956d6..084c0a3b804b 100644
---
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/CliConnectorTransport.java
+++
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/CliConnectorTransport.java
@@ -41,7 +41,8 @@ public interface CliConnectorTransport extends Service {
CamelContext camelContext, CliActionDispatcher dispatcher,
CliSnapshotProducer snapshots, Runnable shutdown);
/**
- * Changes how often the transport polls for actions and updates
snapshots, in millis.
+ * Changes how often the transport updates its snapshots, in millis. The
file transport looks for actions every 100
+ * millis also when this is longer.
*/
default void updateDelay(int delay) {
// noop
diff --git
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ConnectorTasks.java
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ConnectorTasks.java
index 40aa283c3809..a59a9ff7dc12 100644
---
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ConnectorTasks.java
+++
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ConnectorTasks.java
@@ -30,7 +30,7 @@ import org.slf4j.LoggerFactory;
final class ConnectorTasks {
private static final Logger LOG =
LoggerFactory.getLogger(ConnectorTasks.class);
- // an Error is logged as a warning once: the task runs every second, and
may fail every time
+ // an Error is logged as a warning once: the tasks run every second or
more often, and may fail every time
private static final AtomicBoolean WARNED = new AtomicBoolean();
private ConnectorTasks() {
diff --git
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java
index 3bad67b2eb86..7aeb95c64214 100644
---
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java
+++
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java
@@ -46,6 +46,9 @@ import org.slf4j.LoggerFactory;
* to <tt>{pid}-output*.json</tt>, and the snapshots to
<tt>{pid}-status.json</tt>, <tt>{pid}-trace.json</tt> and so on.
* Deleting the <tt>{pid}</tt> lock file shuts down the integration.
* <p/>
+ * The snapshots are written every second (every 100 millis in debug mode),
and the action files are looked for about
+ * every 100 millis in between, on the same thread, so an action does not wait
up to a second for its answer.
+ * <p/>
* An action with a request id (<tt>{pid}-action-{requestId}.json</tt>) always
gets a reply in
* <tt>{pid}-output-{requestId}.json</tt>:
<tt>{"status":"failed","error":"..."}</tt> when it is unknown, throws or
* reports an error (with <tt>"reason":"not-found"</tt> when what it was asked
to work on does not exist, and the result
@@ -66,6 +69,7 @@ public class FileCliConnectorTransport extends ServiceSupport
implements CliConn
private ScheduledExecutorService executor;
private volatile Thread pollThread;
private ScheduledFuture<?> scheduledFuture;
+ private ScheduledFuture<?> actionFuture;
private int delay = 1000;
private long counter;
private final Map<File, PendingAction> pending = new HashMap<>();
@@ -74,6 +78,8 @@ public class FileCliConnectorTransport extends ServiceSupport
implements CliConn
private static final long UNREADABLE_GRACE_MS = 2000;
// how long a stop waits for the poll in flight before it removes the files
private static final long STOP_POLL_TIMEOUT_MS = 5000;
+ // how often the action files are looked for between two snapshots
+ private static final int ACTION_POLL_MS = 100;
private record PendingAction(File output, boolean replyAlways,
CompletableFuture<CliActionOutcome> outcome) {
}
@@ -133,8 +139,7 @@ public class FileCliConnectorTransport extends
ServiceSupport implements CliConn
debugFile = createLockFile(lockFile.getName() + "-debug.json");
receiveFile = createLockFile(lockFile.getName() + "-receive.json");
timelineFile = createLockFile(lockFile.getName() +
"-timeline.json");
- scheduledFuture = executor.scheduleWithFixedDelay(() ->
ConnectorTasks.safely(this::task), 0, delay,
- TimeUnit.MILLISECONDS);
+ schedule();
LOG.info("Camel CLI connector enabled");
} else {
LOG.warn("Cannot create PID file: {}. This integration cannot be
managed by Camel CLI connector.", getPid());
@@ -153,21 +158,34 @@ public class FileCliConnectorTransport extends
ServiceSupport implements CliConn
// ignore
}
}
+ if (actionFuture != null) {
+ actionFuture.cancel(false);
+ actionFuture = null;
+ }
boolean done = scheduledFuture == null || scheduledFuture.isDone();
if (done) {
this.delay = delay;
- scheduledFuture = executor.scheduleWithFixedDelay(() ->
ConnectorTasks.safely(this::task), 0, delay,
- TimeUnit.MILLISECONDS);
+ if (lockFile != null) {
+ schedule();
+ }
}
}
- protected void task() {
- if (!lockFile.exists() && terminating.compareAndSet(false, true)) {
- // if the lock file is deleted then trigger termination
- shutdown.run();
- return;
+ /**
+ * Schedules the poll, and the action poll between two polls when the poll
is slower than it (CAMEL-25447). Both on
+ * the single thread of the executor, so the actions still run one at a
time and in order.
+ */
+ private void schedule() {
+ scheduledFuture = executor.scheduleWithFixedDelay(() ->
ConnectorTasks.safely(this::task), 0, delay,
+ TimeUnit.MILLISECONDS);
+ if (delay > ACTION_POLL_MS) {
+ actionFuture = executor.scheduleWithFixedDelay(() ->
ConnectorTasks.safely(this::actionPollTask),
+ ACTION_POLL_MS, ACTION_POLL_MS, TimeUnit.MILLISECONDS);
}
- if (!statusFile.exists()) {
+ }
+
+ protected void task() {
+ if (!polling()) {
return;
}
@@ -180,6 +198,28 @@ public class FileCliConnectorTransport extends
ServiceSupport implements CliConn
}
}
+ /**
+ * Looks for actions between two snapshots, so an action is answered in
about 100 millis instead of up to a second.
+ */
+ protected void actionPollTask() {
+ if (polling()) {
+ actionTask();
+ }
+ }
+
+ /**
+ * Whether to poll: not when the status file is gone, nor once the lock
file is deleted, which shuts down the
+ * integration.
+ */
+ private boolean polling() {
+ if (!lockFile.exists() && terminating.compareAndSet(false, true)) {
+ // if the lock file is deleted then trigger termination
+ shutdown.run();
+ return false;
+ }
+ return statusFile.exists();
+ }
+
protected synchronized void actionTask() {
pending.entrySet().removeIf(entry -> {
File request = entry.getKey();
@@ -448,6 +488,9 @@ public class FileCliConnectorTransport extends
ServiceSupport implements CliConn
if (scheduledFuture != null) {
scheduledFuture.cancel(false);
}
+ if (actionFuture != null) {
+ actionFuture.cancel(false);
+ }
awaitPoll();
synchronized (this) {
pending.forEach((request, action) -> {
diff --git
a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/FileCliConnectorTransportTest.java
b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/FileCliConnectorTransportTest.java
index 52a953210919..463fe793fdd5 100644
---
a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/FileCliConnectorTransportTest.java
+++
b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/FileCliConnectorTransportTest.java
@@ -22,6 +22,7 @@ import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.util.json.JsonObject;
@@ -102,6 +103,26 @@ class FileCliConnectorTransportTest {
assertThat(awaitJson(file(PID +
"-output-r1.json"))).containsEntry("status", "success");
}
+ @Test
+ void actionsAreAnsweredBetweenTwoStatusUpdates() {
+ // CAMEL-25447: the status every second, as by default, and the
actions looked for about every 100 millis
+ transport.updateDelay(1000);
+ awaitJson(file(PID + "-status.json"));
+ int statuses = snapshots.statuses.get();
+ long start = System.nanoTime();
+
+ for (int i = 1; i <= 5; i++) {
+ writeAction(action("result"), "r" + i);
+ assertThat(awaitJson(file(PID + "-output-r" + i +
".json"))).containsEntry("value", "42");
+ }
+
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
start);
+ // an action per status update took at least 4 seconds
+ assertThat(elapsed).isLessThan(2500);
+ // and the status is not collected more often
+ assertThat(snapshots.statuses.get() -
statuses).isLessThanOrEqualTo((int) (elapsed / 1000) + 2);
+ }
+
@Test
void resultIsTheReply() {
writeAction(action("result"), "r1");
@@ -343,9 +364,11 @@ class FileCliConnectorTransportTest {
private static final class EmptySnapshots implements CliSnapshotProducer {
// thrown once by the next status
volatile Error error;
+ final AtomicInteger statuses = new AtomicInteger();
@Override
public JsonObject status() {
+ statuses.incrementAndGet();
Error e = error;
if (e != null) {
error = null;
