Croway commented on code in PR #27958:
URL: https://github.com/apache/camel/pull/27958#discussion_r4302646062
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java:
##########
@@ -123,6 +132,68 @@ public String stopApplication(long pid) {
return RuntimeHelper.stopApplication(pid);
}
+ /**
+ * The size of the file the CLI connector appends the messages of the
receive console to
+ * ({@code {pid}-receive.json}, read by {@code camel cmd receive} too), to
read only what arrives after this point.
+ */
+ public long receiveFileSize(long pid) {
+ try {
+ Path file = receiveFile(pid);
+ return Files.exists(file) ? Files.size(file) : 0;
+ } catch (IOException e) {
+ return 0;
+ }
+ }
+
+ /**
+ * The messages the CLI connector appended to the receive file after the
given offset: each complete line is a
+ * {@code {"messages": [...]}} dump of the receive console. Only what was
appended after the offset is read, as the
+ * file is append-only and kept across sessions.
+ */
+ public List<JsonObject> readReceivedMessages(long pid, long offset) {
+ List<JsonObject> messages = new ArrayList<>();
+ Path file = receiveFile(pid);
+ try {
+ if (!Files.exists(file)) {
+ return messages;
+ }
+ String text;
+ try (SeekableByteChannel channel = Files.newByteChannel(file,
StandardOpenOption.READ)) {
+ long size = channel.size();
+ // camel cmd receive --action=clear rewrites the file: read it
from the start then
+ long from = offset <= size ? offset : 0;
+ ByteBuffer buffer = ByteBuffer.allocate(Math.toIntExact(size -
from));
Review Comment:
Kept as it is: `offset` is the size of the file at the start of the call (or
0 after a clear, which truncates the file), so `size - from` is what the
connector appended during one call of at most 60 seconds, and it cannot reach 2
GB. If it ever did, a loud `ArithmeticException` is better than a poll loop
that silently finds nothing until the timeout.
_Claude Code on behalf of Croway_
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java:
##########
@@ -662,50 +715,161 @@ private JsonObject doDualJfrRecording(long pid, String
duration, String stacktra
});
}
+ /**
+ * Waits until the recording stopped by itself: sleeps for the time the
console says is left, then checks again (a
+ * recording ends a little after its duration, as it is parsed).
+ */
private void waitForJfrRecordingComplete(long pid, int dur) {
long deadline = System.currentTimeMillis() + (dur + 30) * 1000L;
while (System.currentTimeMillis() < deadline) {
+ JsonObject status = runtimeService.executeAction(pid,
"jfr-memory-leak", root -> {
+ root.put("command", "status");
+ });
+ if (status == null ||
!"recording".equals(status.getString("status"))) {
+ return;
+ }
+ long remaining = status.getLongOrDefault("remainingMs", 1000L);
+ long wait = Math.min(Math.max(remaining, 500L), Math.max(1L,
deadline - System.currentTimeMillis()));
try {
- Thread.sleep(3000);
+ TimeUnit.MILLISECONDS.sleep(wait);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ToolCallException("Interrupted while waiting for
recording", null);
}
- JsonObject status = runtimeService.executeAction(pid,
"jfr-memory-leak", root -> {
- root.put("command", "status");
- });
- if (status != null &&
!"recording".equals(status.getString("status"))) {
- break;
- }
}
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
description = """
- Get the message history trace of the last completed
exchange. \
- This is always captured (no need to enable tracing) and
shows the single most recent exchange \
- with its route path, processors visited, headers, body, and
timing.""")
+ Get the message history trace of the last completed
exchange, with its route path, processors \
+ visited, headers, body, and timing. Recorded when message
history is on and the tracer is \
+ enabled or on standby: the default of camel run (dev
profile); otherwise set \
+ camel.main.message-history=true and
camel.trace.standby=true.""")
public JsonObject camel_runtime_history(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid) {
return delegateToRegistry("get_history", nameOrPid, Map.of());
}
- @Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = false, openWorldHint = false),
+ @Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = true, openWorldHint = false),
description = """
- Receive (poll) a message from a Camel endpoint in the
running application. \
- This is the complement to camel_runtime_send — it consumes
one message from the endpoint.""")
+ Receive (consume) a message from an endpoint of the running
application, the complement of \
+ camel_runtime_send: starts a consumer on the endpoint, waits
for a message, returns it and stops \
+ the consumer again. Receives one message by default; limit
returns up to that many of the messages \
+ that arrived together, and the answer says how many more
were taken and not returned. The consumer \
+ competes with the routes that consume the endpoint, so the
messages it takes do not reach them (a \
+ queue or topic message is gone). To look at queued messages
without taking them use \
+ camel_runtime_browse.""")
public JsonObject camel_runtime_receive(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "Endpoint URI to receive from") String
endpoint) {
+ @ToolArg(description = "Endpoint URI to receive from, or a pattern
matching an endpoint the application "
+ + "sends to, e.g. seda:orders or kafka*")
String endpoint,
+ @ToolArg(description = "How long to wait for a message in seconds
(default 10, at most 60)",
+ required = false) Integer timeoutSeconds,
+ @ToolArg(description = "How many of the messages that arrived
together to return (default 1, at least 1)",
+ required = false) Integer limit) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
- // receive is MCP-only (no ToolRegistry equivalent) — call
RuntimeService directly
+ if (limit != null && limit < 1) {
+ throw new ToolCallException("limit must be at least 1: " + limit,
null);
+ }
+ int timeout = timeoutSeconds == null || timeoutSeconds <= 0
+ ? RECEIVE_DEFAULT_TIMEOUT : Math.min(timeoutSeconds,
RECEIVE_MAX_TIMEOUT);
RuntimeService.ProcessInfo p =
runtimeService.findSingleProcess(nameOrPid);
- return runtimeService.executeAction(p.pid(), "receive", root -> {
+ return receive(p.pid(), endpoint, timeout * 1000L, limit != null ?
limit : 1);
+ }
+
+ /**
+ * Enables the receive console of the process for the endpoint, waits for
the messages the CLI connector writes to
+ * the receive file, and disables it again whatever happens.
+ */
+ JsonObject receive(long pid, String endpoint, long timeoutMs, int limit) {
+ JsonObject state = runtimeService.readStatusSection(pid, "receive");
+ Collection<?> active = state.getCollection("endpoints");
+ if (Boolean.TRUE.equals(state.getBoolean("enabled")) && active != null
&& !active.isEmpty()) {
+ // disabling would stop those consumers too
+ throw new ToolCallException(
+ "The process is already receiving from " + active
+ + " (camel cmd receive); stop that
first with: camel cmd receive --action=stop",
+ null);
+ }
+ long offset = runtimeService.receiveFileSize(pid);
+ JsonObject started = runtimeService.executeAction(pid, "receive", root
-> {
root.put("enabled", "true");
root.put("endpoint", endpoint);
});
+ JsonObject answer;
+ try {
+ answer = awaitMessages(pid, endpoint, started, offset, timeoutMs,
limit);
+ } catch (RuntimeException e) {
+ try {
+ stopReceiving(pid);
+ } catch (RuntimeException stop) {
+ throw new ToolCallException(
+ e.getMessage() + ". The consumer could not be stopped
(" + stop.getMessage()
+ + "): stop it with: camel cmd
receive --action=stop",
+ e);
Review Comment:
Done in f5e8ae9c6d94: a `message(Throwable)` helper gives the message, or
the exception name when there is none, in both places.
_Claude Code on behalf of Croway_
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java:
##########
@@ -662,50 +715,161 @@ private JsonObject doDualJfrRecording(long pid, String
duration, String stacktra
});
}
+ /**
+ * Waits until the recording stopped by itself: sleeps for the time the
console says is left, then checks again (a
+ * recording ends a little after its duration, as it is parsed).
+ */
private void waitForJfrRecordingComplete(long pid, int dur) {
long deadline = System.currentTimeMillis() + (dur + 30) * 1000L;
while (System.currentTimeMillis() < deadline) {
+ JsonObject status = runtimeService.executeAction(pid,
"jfr-memory-leak", root -> {
+ root.put("command", "status");
+ });
+ if (status == null ||
!"recording".equals(status.getString("status"))) {
+ return;
+ }
+ long remaining = status.getLongOrDefault("remainingMs", 1000L);
+ long wait = Math.min(Math.max(remaining, 500L), Math.max(1L,
deadline - System.currentTimeMillis()));
try {
- Thread.sleep(3000);
+ TimeUnit.MILLISECONDS.sleep(wait);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ToolCallException("Interrupted while waiting for
recording", null);
}
- JsonObject status = runtimeService.executeAction(pid,
"jfr-memory-leak", root -> {
- root.put("command", "status");
- });
- if (status != null &&
!"recording".equals(status.getString("status"))) {
- break;
- }
}
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
description = """
- Get the message history trace of the last completed
exchange. \
- This is always captured (no need to enable tracing) and
shows the single most recent exchange \
- with its route path, processors visited, headers, body, and
timing.""")
+ Get the message history trace of the last completed
exchange, with its route path, processors \
+ visited, headers, body, and timing. Recorded when message
history is on and the tracer is \
+ enabled or on standby: the default of camel run (dev
profile); otherwise set \
+ camel.main.message-history=true and
camel.trace.standby=true.""")
public JsonObject camel_runtime_history(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid) {
return delegateToRegistry("get_history", nameOrPid, Map.of());
}
- @Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = false, openWorldHint = false),
+ @Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = true, openWorldHint = false),
description = """
- Receive (poll) a message from a Camel endpoint in the
running application. \
- This is the complement to camel_runtime_send — it consumes
one message from the endpoint.""")
+ Receive (consume) a message from an endpoint of the running
application, the complement of \
+ camel_runtime_send: starts a consumer on the endpoint, waits
for a message, returns it and stops \
+ the consumer again. Receives one message by default; limit
returns up to that many of the messages \
+ that arrived together, and the answer says how many more
were taken and not returned. The consumer \
+ competes with the routes that consume the endpoint, so the
messages it takes do not reach them (a \
+ queue or topic message is gone). To look at queued messages
without taking them use \
+ camel_runtime_browse.""")
public JsonObject camel_runtime_receive(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "Endpoint URI to receive from") String
endpoint) {
+ @ToolArg(description = "Endpoint URI to receive from, or a pattern
matching an endpoint the application "
+ + "sends to, e.g. seda:orders or kafka*")
String endpoint,
+ @ToolArg(description = "How long to wait for a message in seconds
(default 10, at most 60)",
+ required = false) Integer timeoutSeconds,
+ @ToolArg(description = "How many of the messages that arrived
together to return (default 1, at least 1)",
+ required = false) Integer limit) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
- // receive is MCP-only (no ToolRegistry equivalent) — call
RuntimeService directly
+ if (limit != null && limit < 1) {
+ throw new ToolCallException("limit must be at least 1: " + limit,
null);
+ }
+ int timeout = timeoutSeconds == null || timeoutSeconds <= 0
+ ? RECEIVE_DEFAULT_TIMEOUT : Math.min(timeoutSeconds,
RECEIVE_MAX_TIMEOUT);
RuntimeService.ProcessInfo p =
runtimeService.findSingleProcess(nameOrPid);
- return runtimeService.executeAction(p.pid(), "receive", root -> {
+ return receive(p.pid(), endpoint, timeout * 1000L, limit != null ?
limit : 1);
+ }
+
+ /**
+ * Enables the receive console of the process for the endpoint, waits for
the messages the CLI connector writes to
+ * the receive file, and disables it again whatever happens.
+ */
+ JsonObject receive(long pid, String endpoint, long timeoutMs, int limit) {
+ JsonObject state = runtimeService.readStatusSection(pid, "receive");
+ Collection<?> active = state.getCollection("endpoints");
+ if (Boolean.TRUE.equals(state.getBoolean("enabled")) && active != null
&& !active.isEmpty()) {
+ // disabling would stop those consumers too
+ throw new ToolCallException(
+ "The process is already receiving from " + active
+ + " (camel cmd receive); stop that
first with: camel cmd receive --action=stop",
+ null);
+ }
+ long offset = runtimeService.receiveFileSize(pid);
+ JsonObject started = runtimeService.executeAction(pid, "receive", root
-> {
root.put("enabled", "true");
root.put("endpoint", endpoint);
});
+ JsonObject answer;
+ try {
+ answer = awaitMessages(pid, endpoint, started, offset, timeoutMs,
limit);
+ } catch (RuntimeException e) {
+ try {
+ stopReceiving(pid);
+ } catch (RuntimeException stop) {
+ throw new ToolCallException(
+ e.getMessage() + ". The consumer could not be stopped
(" + stop.getMessage()
+ + "): stop it with: camel cmd
receive --action=stop",
+ e);
+ }
+ throw e;
+ }
+ try {
+ stopReceiving(pid);
+ } catch (RuntimeException e) {
+ // the messages are taken from the endpoint already: return them,
and say the consumer may still run
+ answer.put("warning", "The consumer could not be stopped (" +
e.getMessage()
+ + ") and may still take messages from the
endpoint: stop it with: "
+ + "camel cmd receive --action=stop");
+ }
+ return answer;
+ }
+
+ private JsonObject awaitMessages(
+ long pid, String endpoint, JsonObject started, long offset, long
timeoutMs, int limit) {
+ if (started.getString("error") != null) {
+ throw new ToolCallException("Cannot receive from " + endpoint + ":
" + started.getString("error"), null);
+ }
+ String uri = started.getString("url");
+ if (uri == null) {
+ throw new ToolCallException(
+ "No endpoint matches " + endpoint
+ + " (a pattern matches the endpoints
the application sends to; "
+ + "camel_runtime_endpoints lists
them)",
+ null);
+ }
+ JsonObject answer = new JsonObject();
+ answer.put("endpoint", uri);
+ StopWatch watch = new StopWatch();
+ while (true) {
+ List<JsonObject> messages =
runtimeService.readReceivedMessages(pid, offset);
Review Comment:
Done in f5e8ae9c6d94, with the status read after the stop too: `receive`
reads `total` from the receive status before enabling and keeps only the
messages with a `uid` above it; after the stop it waits (3 s at most, the
connector applies the stop and writes the status in one tick and dumps the
queue every second tick) until the status shows the console disabled, and the
`total` then is the exact number this call took, so the messages dumped after
the stop are returned as well (up to `limit`) and otherwise counted in
`notReturned`. When the process does not confirm the stop or has not written
every message in that time, the note says so. A `ReentrantLock` per pid runs
one call at a time: a second call waits up to its own timeout, then fails with
a message naming the first. Tests in `RuntimeToolsWiringTest` with a recording
runtime whose status follows the actions (leftovers of an earlier session,
messages written after the stop, unwritten and unconfirmed cases, parallel
call).
_Claude Code on behalf of Croway_
--
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]
