gnodet-bot commented on code in PR #27958:
URL: https://github.com/apache/camel/pull/27958#discussion_r4300252324
##########
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:
💡 **Robustness (low):** `e.getMessage()` and `stop.getMessage()` can both
return `null` (e.g. when the exception is constructed without a message, or is
a `NullPointerException`). Java string concatenation coerces `null` to the
literal string `"null"`, producing a confusing message like `"null. The
consumer could not be stopped (null): stop it with: camel cmd receive
--action=stop"`.
Consider:
```suggestion
throw new ToolCallException(
e + ". The consumer could not be stopped (" + stop
+ "): stop it with: camel cmd
receive --action=stop",
e);
```
`Throwable.toString()` always returns a non-null string (e.g.
`java.lang.RuntimeException: message` or just
`java.lang.NullPointerException`), giving more context than `getMessage()`
alone.
##########
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:
⚠️ **Bug (medium):** `Math.toIntExact(size - from)` throws
`ArithmeticException` (extends `RuntimeException`) on overflow — not
`IOException`. The `catch (IOException e)` block below (line 187) won't catch
it. If the receive file ever grows beyond ~2 GB, this exception escapes
unhandled.
The comment says `// not readable yet: the caller polls again`, implying any
failure here should be swallowed. Add a guard:
```suggestion
if (size - from > Integer.MAX_VALUE) {
return messages; // file too large to read in one
buffer; caller polls again
}
ByteBuffer buffer = ByteBuffer.allocate((int) (size - from));
```
Or, if you prefer keeping `Math.toIntExact`, wrap it:
```java
} catch (IOException | ArithmeticException e) {
// not readable yet: the caller polls again
}
```
--
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]
