davsclaus commented on code in PR #27958:
URL: https://github.com/apache/camel/pull/27958#discussion_r4301786607
##########
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:
Following up on @allthingssecurity's review: this returns every message
appended after `offset`, also the leftovers of an earlier call (taken after the
dump it read, written on the next tick) and those of a parallel call on the
same process. Reading the console's `total` from the `receive` status section
before enabling, and keeping only messages with a `uid` above it, makes the
answer this call's own:
```java
long startTotal = state.getLongOrDefault("total", 0L); // in receive(),
before enabling
...
messages.removeIf(m -> m.getLongOrDefault("uid", 0L) <= startTotal);
```
The undercount (messages taken in the last tick before the stop) needs the
status read after the stop has been applied; if that is too much for this PR,
the note at line 852 ("taken from the endpoint all the same") could say that
more messages than counted may have been taken, so the agent does not take
`received` as the exact number.
--
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]
