Croway commented on code in PR #27958:
URL: https://github.com/apache/camel/pull/27958#discussion_r4300210732
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java:
##########
@@ -123,6 +129,58 @@ 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 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.
+ */
+ public List<JsonObject> readReceivedMessages(long pid, long offset) {
+ List<JsonObject> messages = new ArrayList<>();
+ Path file = receiveFile(pid);
+ try {
+ if (!Files.exists(file)) {
+ return messages;
+ }
+ byte[] data = Files.readAllBytes(file);
Review Comment:
Done in 6676883e9f13: `readReceivedMessages` opens a `SeekableByteChannel`,
positions it at the offset and reads only `size - offset` bytes (from 0 when
the file is shorter than the offset, after a clear). The clear case is added to
`readReceivedMessagesReadsTheLinesAfterTheOffset`.
_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 +727,120 @@ 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.main.tracing-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. The consumer competes with the routes
that consume the endpoint, so the \
+ message it takes does 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) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
- // receive is MCP-only (no ToolRegistry equivalent) — call
RuntimeService directly
+ 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);
+ }
+
+ /**
+ * 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) {
+ 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 receive); stop that first
with: camel receive --action=stop",
+ null);
+ }
+ long offset = runtimeService.receiveFileSize(pid);
+ JsonObject started = runtimeService.executeAction(pid, "receive", root
-> {
root.put("enabled", "true");
root.put("endpoint", endpoint);
});
+ try {
+ 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);
+ if (!messages.isEmpty()) {
+ answer.put("received", messages.size());
+ answer.put("messages", messages);
+ return answer;
+ }
+ if (watch.taken() >= timeoutMs) {
+ answer.put("received", 0);
+ answer.put("note", "No message arrived within " +
(timeoutMs / 1000) + " seconds");
+ return answer;
+ }
+ try {
+ TimeUnit.MILLISECONDS.sleep(RECEIVE_POLL_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ToolCallException("Interrupted while waiting for
a message", null);
+ }
+ }
+ } finally {
+ // an action without an endpoint stops the consumers of the
receive console
+ runtimeService.executeAction(pid, "receive", null);
Review Comment:
Done in 6676883e9f13: the `finally` is gone; the wait is `awaitMessages` and
the stop is `stopReceiving`. When the stop fails after messages arrived, they
are returned with a `warning` that names `camel cmd receive --action=stop`;
when the receive itself failed and the stop fails too, the error says the
consumer could not be stopped and names the command. Tests
`receiveReturnsTheMessageWithAWarningWhenTheConsumerCannotBeStopped` and
`receiveNamesTheStopCommandWhenItFailsAndCannotStopTheConsumer`.
_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]
