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 4209955441d8 CAMEL-26767: camel-jbang - camel cmd send: --repeat,
--interval and a folder of messages, to feed a route (#29459)
4209955441d8 is described below
commit 4209955441d84c4d0bbf35c8b208db0ba2cda874
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 9 14:07:01 2026 +0100
CAMEL-26767: camel-jbang - camel cmd send: --repeat, --interval and a
folder of messages, to feed a route (#29459)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/impl/console/SendDevConsole.java | 12 +-
.../camel/impl/console/SendDevConsoleTest.java | 25 +++
.../modules/ROOT/pages/camel-jbang-devtools.adoc | 32 +++
.../pages/jbang-commands/camel-jbang-cmd-send.adoc | 2 +
.../camel/cli/connector/LocalCliConnector.java | 139 +++++++++---
.../apache/camel/cli/connector/RepeatedSend.java | 240 +++++++++++++++++++++
.../connector/LocalCliConnectorSendActionTest.java | 228 ++++++++++++++++++++
.../META-INF/camel-jbang-commands-metadata.json | 2 +-
.../core/commands/action/CamelSendAction.java | 123 ++++++++++-
.../dsl/jbang/core/commands/ai/ToolContext.java | 6 +-
.../dsl/jbang/core/commands/ai/ToolRegistry.java | 38 +++-
.../camel/dsl/jbang/core/common/RuntimeHelper.java | 43 +++-
.../commands/action/CamelSendActionResultTest.java | 73 +++++++
.../jbang/core/commands/ai/RuntimeToolsTest.java | 11 +
.../dsl/jbang/core/common/RuntimeHelperTest.java | 17 ++
.../src/test/resources/golden/cli-help.txt | 26 ++-
.../dsl/jbang/core/commands/mcp/RuntimeTools.java | 22 +-
.../jbang/core/commands/mcp/RuntimeToolsTest.java | 2 +-
.../core/commands/mcp/RuntimeToolsWiringTest.java | 3 +-
19 files changed, 981 insertions(+), 63 deletions(-)
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
index b015c30b303f..cc83ced9e0f1 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
@@ -16,9 +16,8 @@
*/
package org.apache.camel.impl.console;
-import java.io.Closeable;
import java.io.File;
-import java.io.FileInputStream;
+import java.nio.file.Files;
import java.util.Base64;
import java.util.HashMap;
import java.util.List;
@@ -39,7 +38,6 @@ import org.apache.camel.support.ExceptionHelper;
import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.console.AbstractDevConsole;
import org.apache.camel.support.service.ServiceHelper;
-import org.apache.camel.util.IOHelper;
import org.apache.camel.util.StopWatch;
import org.apache.camel.util.TimeUtils;
import org.apache.camel.util.json.JsonRecordSupport;
@@ -282,9 +280,6 @@ public class SendDevConsole extends AbstractDevConsole {
"InOut".equals(mep) ? ExchangePattern.InOut :
ExchangePattern.InOnly);
});
}
- if (inputBody instanceof Closeable c) {
- IOHelper.close(c);
- }
}
return out;
}
@@ -339,8 +334,9 @@ public class SendDevConsole extends AbstractDevConsole {
}
Object b = body;
if (body.startsWith("file:")) {
- File file = new File(body.substring(5));
- b = new FileInputStream(file);
+ // the content of the file, not a stream: an endpoint that hands
the message to another thread (seda, a
+ // stub) reads the body after the send returns, when a stream
would be closed already
+ b = Files.readAllBytes(new File(body.substring(5)).toPath());
}
return b;
}
diff --git
a/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
b/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
index acbc85970291..331046d4bbda 100644
---
a/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
+++
b/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
@@ -17,6 +17,8 @@
package org.apache.camel.impl.console;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
@@ -29,6 +31,7 @@ import org.apache.camel.console.DevConsole;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.json.JsonObject;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -41,6 +44,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public class SendDevConsoleTest extends AbstractDevConsoleTest {
+ @TempDir
+ Path dir;
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -131,6 +137,25 @@ public class SendDevConsoleTest extends
AbstractDevConsoleTest {
assertEquals("error", jsonOut.getString("status"));
}
+ @Test
+ public void testSendConsoleFileBodyToAQueue() throws Exception {
+ // seda hands the message to another thread, which reads the body
after the send returns
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:refunds").delay(200).to("mock:refunds");
+ }
+ });
+ Path file = Files.writeString(dir.resolve("refund.json"),
"{\"refund\": 1}");
+ MockEndpoint mock = getMockEndpoint("mock:refunds");
+ mock.expectedBodiesReceived("{\"refund\": 1}");
+
+ JsonObject jsonOut = callJson(startedSendConsole(),
createOptions("seda:refunds", "file:" + file));
+
+ assertEquals("success", jsonOut.getString("status"), jsonOut.toJson());
+ mock.assertIsSatisfied();
+ }
+
@Test
public void testSendConsoleGettersSetters() {
SendDevConsole console = new SendDevConsole();
diff --git a/docs/user-manual/modules/ROOT/pages/camel-jbang-devtools.adoc
b/docs/user-manual/modules/ROOT/pages/camel-jbang-devtools.adoc
index 6cb4838dbd7b..6ca1d495ee63 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-jbang-devtools.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-devtools.adoc
@@ -57,6 +57,38 @@ $ camel cmd send mqtt --body=file:payload.json
--endpoint=mqtt
TIP: Run `camel cmd send --help` for all options. Full example on
https://github.com/apache/camel-jbang-examples/tree/main/mqtt[GitHub].
+=== Sending many messages
+
+*Available since Camel 4.24*
+
+`--repeat` sends the message that many times, with `--interval` between two
(in millis, or a duration such as
+`200ms`), and prints a summary instead of each reply. It exits with `1` when a
message failed. A range such as
+`--interval=100..500` (or `100ms..2s`, as the `duration=` of a chaos rule)
draws each pause at random, so the messages
+do not come at a steady pace:
+
+[source,bash]
+----
+$ camel cmd send shop --endpoint=kafka:refunds --body=file:refund.json
--repeat=50 --interval=200ms
+2026-11-09 12:01:14.101 58473 --- kafka://refunds :
Sent 50 messages (48 ok, 2 failed) (10s)
+ 2 x IllegalStateException: Refund amount is missing
+----
+
+A body of `file:<folder>` sends the files of the folder in name order, one
message per file. A `.properties` file
+with the same name has the headers of a message: `001.properties` for
`001.json`. With `--repeat` it goes round the
+files again, and the headers of `--header` go with every message:
+
+[source,bash]
+----
+$ ls test/chaos/refunds-fails
+001.json 001.properties 002.json 003.json
+$ camel cmd send shop --endpoint=kafka:refunds
--body=file:test/chaos/refunds-fails/ --interval=100ms
+----
+
+The integration reads the files, as it reads the file of a single body, and
sends the messages on a thread of its own,
+so `camel get` and the other commands keep working meanwhile. At most 10000
messages are sent at a time, with at most
+60 seconds between two. A broker consumer stubbed with `camel run
--stub=remote` is fed this way, which is how chaos
+rules can be tried on it (see
xref:next@components:others:chaos.adoc[camel-chaos]).
+
=== Sending without a running integration
*Available since Camel 4.12*
diff --git
a/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-cmd-send.adoc
b/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-cmd-send.adoc
index 18b33ecf00e9..c111a03b94d8 100644
---
a/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-cmd-send.adoc
+++
b/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-cmd-send.adoc
@@ -25,11 +25,13 @@ camel cmd send [options]
| `--fresh` | Make sure we use fresh (i.e. non-cached) resources | false |
boolean
| `--header` | Message header (key=value) | | List
| `--infra` | Send to infrastructure service (e.g., nats, kafka) | | String
+| `--interval` | The pause between two messages with --repeat or a folder
body, in millis or as a duration such as 200ms or 1s, or a range such as
100..500 or 100ms..2s to draw each pause at random (at most 60s) | | String
| `--logging-color` | Use colored logging | true | boolean
| `--poll` | Poll instead of sending a message. This can be used to receive
latest message from a Kafka topic or JMS queue. | | boolean
| `--pretty` | Pretty print response message body (InOut) when using JSon or
XML format | | boolean
| `--prop,--property` | Additional properties; override existing (only
applicable when NOT using an existing running Camel) | | String[]
| `--properties` | comma separated list of properties file (only applicable
when NOT using an existing running Camel) (ex.
/path/to/file.properties,/path/to/other.properties | | String
+| `--repeat` | Send the message this many times (going round the files of a
folder body), and print a summary instead of each reply (at most 10000) | | int
| `--reply` | Whether to expect a reply message (InOut vs InOut messaging
style) | | boolean
| `--reply-file` | Saves reply message to the file with the given name
(override if exists) | | String
| `--repo,--repos` | Additional maven repositories for download on-demand (Use
commas to separate multiple repositories) | | String
diff --git
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java
index d008e364d619..ef4d34f672d2 100644
---
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java
+++
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java
@@ -129,6 +129,7 @@ public class LocalCliConnector extends ServiceSupport
private volatile ExecutorService terminateExecutor;
private ExecutorService semanticExecutor;
private ExecutorService auditExecutor;
+ private ExecutorService sendExecutor;
private final AtomicLong semanticConsoleLookup = new
AtomicLong(System.nanoTime() - SEMANTIC_CONSOLE_RETRY_NANOS);
private ProducerTemplate producer;
private ConsumerTemplate consumer;
@@ -244,6 +245,9 @@ public class LocalCliConnector extends ServiceSupport
@Override
public CompletableFuture<Boolean> dispatchAsync(JsonObject root,
CliActionOutput output) {
+ if ("send".equals(root.getString("action")) &&
RepeatedSend.isRepeated(root)) {
+ return repeatedSend(root, output);
+ }
boolean auditQuery = "semantic-audit".equals(root.getString("action"));
if (!auditQuery &&
!"semantic-evaluate".equals(root.getString("action"))) {
return CliActionDispatcher.super.dispatchAsync(root, output);
@@ -292,6 +296,64 @@ public class LocalCliConnector extends ServiceSupport
return result;
}
+ /**
+ * The send action with a repeat or a folder of messages, on a thread of
its own: it may take a while, and the
+ * status and the other actions go on meanwhile. The transport cancels it
when the client gives up the request.
+ */
+ private CompletableFuture<Boolean> repeatedSend(JsonObject root,
CliActionOutput output) {
+ CompletableFuture<Boolean> result = new CompletableFuture<>();
+ FutureTask<Void> task = new FutureTask<>(() -> {
+ try {
+ // capture this request's output; never use the dispatcher's
mutable actionOutput from a worker
+ output.write(repeatedSendResult(root));
+ result.complete(true);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ result.cancel(true);
+ } catch (Exception e) {
+ result.completeExceptionally(e);
+ }
+ return null;
+ });
+ result.whenComplete((value, failure) -> {
+ if (failure != null) {
+ task.cancel(true);
+ }
+ });
+ try {
+ synchronized (this) {
+ if (isStoppingOrStopped()) {
+ throw new IllegalStateException("CLI connector is stopping
or stopped");
+ }
+ if (sendExecutor == null) {
+ sendExecutor = new
ThreadPoolBuilder(camelContext).poolSize(2).maxPoolSize(2).maxQueueSize(0)
+
.rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build("cli-send",
"CliRepeatedSend");
+ }
+ sendExecutor.execute(task);
+ }
+ } catch (RejectedExecutionException busy) {
+ result.completeExceptionally(new IllegalStateException("Busy: two
repeated sends are already running"));
+ } catch (Exception e) {
+ result.completeExceptionally(e);
+ }
+ return result;
+ }
+
+ private JsonObject repeatedSendResult(JsonObject root) throws Exception {
+ DevConsole dc =
camelContext.getCamelContextExtension().getContextPlugin(DevConsoleRegistry.class)
+ .resolveById("send");
+ if (dc == null) {
+ return new JsonObject();
+ }
+ if (root.getBooleanOrDefault("poll", false)) {
+ throw new IllegalArgumentException("A poll does not repeat: send
with repeat, or poll once");
+ }
+ waitUntilStarted();
+ return RepeatedSend.send(dc, sendOptions(root), sendHeaders(root),
root.getString("body"),
+ root.getLongOrDefault("repeat", 0),
root.getLongOrDefault("interval", 0),
+ root.getLongOrDefault("intervalMax", 0));
+ }
+
private ExecutorService newSemanticExecutor(String source, String name)
throws Exception {
return new
ThreadPoolBuilder(camelContext).poolSize(2).maxPoolSize(2).maxQueueSize(0)
.rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build(source,
name);
@@ -667,51 +729,58 @@ public class LocalCliConnector extends ServiceSupport
.resolveById("send");
if (dc != null) {
- // wait for camel to be ready
- StopWatch watch = new StopWatch();
- while (!camelContext.isStarted() && watch.taken() < 5000) {
- Thread.sleep(100);
+ if (RepeatedSend.isRepeated(root)) {
+ // a transport that dispatches on its own thread, rather than
with dispatchAsync
+ writeOutput(repeatedSendResult(root));
+ return;
}
+ waitUntilStarted();
- JsonObject json;
- String endpoint = root.getString("endpoint");
+ final Map<String, Object> args = sendOptions(root);
String body = root.getString("body");
- String exchangePattern = root.getString("exchangePattern");
- String poll = root.getString("poll");
- String pollTimeout = root.getString("pollTimeout");
- final Map<String, Object> args = new LinkedHashMap<>();
- if (endpoint != null) {
- args.put("endpoint", endpoint);
- }
if (body != null) {
args.put("body", body);
}
- String bodyEncoding = root.getString("bodyEncoding");
- if (bodyEncoding != null) {
- args.put("bodyEncoding", bodyEncoding);
- }
- if (exchangePattern != null) {
- args.put("exchangePattern", exchangePattern);
- }
- if (poll != null) {
- args.put("poll", poll);
- }
- if (pollTimeout != null) {
- args.put("pollTimeout", pollTimeout);
- }
- Collection<JsonObject> headers = root.getCollection("headers");
- if (headers != null) {
- for (JsonObject jo : headers) {
- args.put(jo.getString("key"), jo.getString("value"));
- }
- }
- json = (JsonObject) dc.call(DevConsole.MediaType.JSON, args);
+ args.putAll(sendHeaders(root));
+ JsonObject json = (JsonObject) dc.call(DevConsole.MediaType.JSON,
args);
writeOutput(json);
} else {
writeOutput(new JsonObject());
}
}
+ private void waitUntilStarted() throws InterruptedException {
+ // wait for camel to be ready
+ StopWatch watch = new StopWatch();
+ while (!camelContext.isStarted() && watch.taken() < 5000) {
+ Thread.sleep(100);
+ }
+ }
+
+ /** The options of the send console from the send action, without the body
and the headers. */
+ private static Map<String, Object> sendOptions(JsonObject root) {
+ final Map<String, Object> args = new LinkedHashMap<>();
+ for (String key : List.of("endpoint", "bodyEncoding",
"exchangePattern", "poll", "pollTimeout")) {
+ String value = root.getString(key);
+ if (value != null) {
+ args.put(key, value);
+ }
+ }
+ return args;
+ }
+
+ /** The headers of the send action, as the send console takes them. */
+ private static Map<String, Object> sendHeaders(JsonObject root) {
+ final Map<String, Object> args = new LinkedHashMap<>();
+ Collection<JsonObject> headers = root.getCollection("headers");
+ if (headers != null) {
+ for (JsonObject jo : headers) {
+ args.put(jo.getString("key"), jo.getString("value"));
+ }
+ }
+ return args;
+ }
+
private void doActionStubTask(JsonObject root) throws Exception {
String filter = root.getString("filter");
String limit = root.getString("limit");
@@ -2015,6 +2084,10 @@ public class LocalCliConnector extends ServiceSupport
camelContext.getExecutorServiceManager().shutdownNow(semanticExecutor);
semanticExecutor = null;
}
+ if (sendExecutor != null) {
+
camelContext.getExecutorServiceManager().shutdownNow(sendExecutor);
+ sendExecutor = null;
+ }
}
ServiceHelper.stopService(producer, consumer);
}
diff --git
a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/RepeatedSend.java
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/RepeatedSend.java
new file mode 100644
index 000000000000..a1956dd53f6d
--- /dev/null
+++
b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/RepeatedSend.java
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.cli.connector;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ThreadLocalRandom;
+
+import org.apache.camel.console.DevConsole;
+import org.apache.camel.util.FileUtil;
+import org.apache.camel.util.TimeUtils;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+
+/**
+ * The send action with a repeat or a folder of messages ({@code camel cmd
send --repeat=50 --interval=200ms}, and
+ * {@code --body=file:<folder>/}): it sends the messages through the send dev
console one after another, with a pause
+ * between them, and answers with a summary when done (CAMEL-26767). It runs
on a thread of its own, so the status and
+ * the other actions of the connector go on meanwhile, and it stops when the
client gives up the request.
+ * <p/>
+ * The messages of a folder are its files in name order, one message per file,
leaving out hidden files and the
+ * {@code .properties} files: {@code 001.properties} has the headers of {@code
001.json}. The folder is read by the
+ * integration, relative to its working directory, as the file of a single
body is.
+ */
+public final class RepeatedSend {
+
+ /** The most messages one send action sends; camel cmd send checks it
before it asks. */
+ public static final int MAX_MESSAGES = 10000;
+ /** The longest pause between two messages, in millis; camel cmd send
checks it before it asks. */
+ public static final long MAX_INTERVAL = 60000;
+ private static final int MAX_ERRORS = 5;
+
+ private RepeatedSend() {
+ }
+
+ /** A message to send: its body, and its own headers (from the properties
file next to it). */
+ record Message(String body, Map<String, Object> headers) {
+ }
+
+ /**
+ * The pause between two messages, as text: a time such as {@code 200},
{@code 200ms} or {@code 1s}, or a range such
+ * as {@code 100..500} or {@code 100ms..2s} (as the {@code duration=} of a
chaos rule), from which each pause is
+ * drawn at random, so the messages do not come at a steady pace.
+ *
+ * @return the shortest and the longest pause in
millis (the same for a time)
+ * @throws IllegalArgumentException when it is not a time or a range, or
longer than {@link #MAX_INTERVAL}
+ */
+ public static long[] interval(String text) {
+ if (text == null || text.isBlank()) {
+ return new long[] { 0, 0 };
+ }
+ String[] parts = text.strip().split("\\.\\.", -1);
+ if (parts.length > 2) {
+ throw new IllegalArgumentException("The interval is not a time or
a range such as 100..500: " + text);
+ }
+ long[] answer = new long[2];
+ try {
+ answer[0] = TimeUtils.toMilliSeconds(parts[0].strip());
+ answer[1] = parts.length == 2 ?
TimeUtils.toMilliSeconds(parts[1].strip()) : answer[0];
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException("The interval is not a time or
a range such as 100..500: " + text);
+ }
+ if (answer[0] < 0 || answer[1] < answer[0]) {
+ throw new IllegalArgumentException("The interval must go from the
shortest to the longest pause: " + text);
+ }
+ if (answer[1] > MAX_INTERVAL) {
+ throw new IllegalArgumentException("The interval must be at most
60s, was: " + text);
+ }
+ return answer;
+ }
+
+ /** Whether the send action repeats, or sends a folder of messages, rather
than one message. */
+ static boolean isRepeated(JsonObject root) {
+ return root.getLongOrDefault("repeat", 0) > 1 ||
folder(root.getString("body")) != null;
+ }
+
+ /** The folder a body names ({@code file:<folder>}), or null when it names
none. */
+ static File folder(String body) {
+ if (body == null || !body.startsWith("file:")) {
+ return null;
+ }
+ File file = new File(body.substring(5));
+ return file.isDirectory() ? file : null;
+ }
+
+ /**
+ * The messages to send: the files of the folder the body names, or the
body itself.
+ *
+ * @throws IllegalArgumentException when the folder has no message
+ */
+ static List<Message> messages(String body) throws IOException {
+ File folder = folder(body);
+ if (folder == null) {
+ return List.of(new Message(body, Map.of()));
+ }
+ File[] files = folder.listFiles(f -> f.isFile() &&
!f.getName().startsWith(".")
+ && !f.getName().endsWith(".properties"));
+ if (files == null || files.length == 0) {
+ throw new IllegalArgumentException("No messages in the folder " +
folder);
+ }
+ Arrays.sort(files, Comparator.comparing(File::getName));
+ List<Message> answer = new ArrayList<>();
+ for (File file : files) {
+ answer.add(new Message("file:" + file.getAbsolutePath(),
headers(file)));
+ }
+ return answer;
+ }
+
+ /** The headers of a message of a folder: the properties file with the
same name, such as 001.properties. */
+ private static Map<String, Object> headers(File file) throws IOException {
+ File props = new File(file.getParentFile(),
FileUtil.stripExt(file.getName()) + ".properties");
+ if (!props.isFile()) {
+ return Map.of();
+ }
+ Properties properties = new Properties();
+ try (InputStream is = Files.newInputStream(props.toPath())) {
+ properties.load(is);
+ }
+ Map<String, Object> answer = new LinkedHashMap<>();
+ properties.stringPropertyNames().stream().sorted().forEach(k ->
answer.put(k, properties.getProperty(k)));
+ return answer;
+ }
+
+ /**
+ * Sends the messages and sums up how it went.
+ *
+ * @param console the send dev console
+ * @param options the options of the send console without
the body and headers (endpoint, exchange
+ * pattern and so on)
+ * @param headers the headers of every message; the headers
of a message of a folder win
+ * @param body the body: a message, a file, or a folder
of messages
+ * @param repeat how many messages to send in all, going
round the messages again; 0 or less to send
+ * each message once
+ * @param interval the pause between two messages, in millis,
or the shortest pause of a range
+ * @param intervalMax the longest pause of a range, from which
each pause is drawn at random; the interval
+ * or less for the same pause each time
+ * @throws InterruptedException when the client gave up the request
+ */
+ static JsonObject send(
+ DevConsole console, Map<String, Object> options, Map<String,
Object> headers, String body, long repeat,
+ long interval, long intervalMax)
+ throws IOException, InterruptedException {
+ if (repeat > MAX_MESSAGES) {
+ throw new IllegalArgumentException("At most " + MAX_MESSAGES + "
messages can be sent at a time, not " + repeat);
+ }
+ long longest = Math.max(interval, intervalMax);
+ if (interval < 0 || longest > MAX_INTERVAL) {
+ throw new IllegalArgumentException("The interval must be between 0
and " + MAX_INTERVAL + " millis");
+ }
+ List<Message> messages = messages(body);
+ long count = repeat > 0 ? repeat : messages.size();
+ long start = System.currentTimeMillis();
+ String endpoint = null;
+ long sent = 0;
+ long failed = 0;
+ Map<String, Long> errors = new LinkedHashMap<>();
+ for (long i = 0; i < count; i++) {
+ if (i > 0 && longest > 0) {
+ Thread.sleep(longest > interval ?
ThreadLocalRandom.current().nextLong(interval, longest + 1) : interval);
+ }
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException("The client gave up the
request");
+ }
+ Message message = messages.get((int) (i % messages.size()));
+ Map<String, Object> args = new LinkedHashMap<>(options);
+ args.putAll(headers);
+ args.putAll(message.headers());
+ if (message.body() != null) {
+ args.put("body", message.body());
+ }
+ JsonObject response = (JsonObject)
console.call(DevConsole.MediaType.JSON, args);
+ if (endpoint == null) {
+ endpoint = response.getString("endpoint");
+ }
+ sent++;
+ if ("success".equals(response.getString("status"))) {
+ continue;
+ }
+ failed++;
+ JsonObject exception = response.getMap("exception");
+ String type = exception != null ? exception.getString("type") :
null;
+ String error = exception != null
+ ? simpleName(type) + ": " + exception.getString("message")
+ : response.getString("status");
+ if (errors.containsKey(error) || errors.size() < MAX_ERRORS) {
+ errors.merge(error, 1L, Long::sum);
+ }
+ if (type != null && type.endsWith("NoSuchEndpointException")) {
+ // nothing will reach an endpoint that is not there
+ break;
+ }
+ }
+ JsonObject answer = new JsonObject();
+ answer.put("timestamp", start);
+ answer.put("status", failed > 0 ? "failed" : "success");
+ answer.put("elapsed", System.currentTimeMillis() - start);
+ if (endpoint != null) {
+ answer.put("endpoint", endpoint);
+ }
+ answer.put("messages", count);
+ answer.put("sent", sent);
+ answer.put("failed", failed);
+ JsonArray list = new JsonArray();
+ errors.forEach((message, n) -> {
+ JsonObject jo = new JsonObject();
+ jo.put("message", message);
+ jo.put("count", n);
+ list.add(jo);
+ });
+ answer.put("errors", list);
+ return answer;
+ }
+
+ private static String simpleName(String type) {
+ return type != null && type.lastIndexOf('.') >= 0 ?
type.substring(type.lastIndexOf('.') + 1) : type;
+ }
+}
diff --git
a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSendActionTest.java
b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSendActionTest.java
new file mode 100644
index 000000000000..3bdb264a73c8
--- /dev/null
+++
b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSendActionTest.java
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.cli.connector;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.CliConnectorFactory;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * The send action with a repeat or a folder of messages (camel cmd send
--repeat --interval, CAMEL-26767): it runs on a
+ * thread of its own and answers with a summary.
+ */
+class LocalCliConnectorSendActionTest {
+
+ @TempDir
+ Path dir;
+
+ private final List<String> received = new CopyOnWriteArrayList<>();
+ private CamelContext context;
+ private LocalCliConnector connector;
+
+ @BeforeEach
+ void start() throws Exception {
+ context = new DefaultCamelContext();
+ // the action is dispatched here: a connector of the context would
poll the {pid} files of other tests
+ var disabled = new DefaultCliConnectorFactory();
+ disabled.setEnabled(false);
+
context.getCamelContextExtension().addContextPlugin(CliConnectorFactory.class,
disabled);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:refunds").routeId("refunds")
+ .process(e ->
received.add(e.getMessage().getBody(String.class) + " "
+ +
e.getMessage().getHeader("country", "-", String.class)))
+ .filter(body().contains("bad")).throwException(new
IllegalStateException("bad refund"));
+ }
+ });
+ context.start();
+ connector = new LocalCliConnector(new DefaultCliConnectorFactory());
+ connector.setCamelContext(context);
+ connector.start();
+ }
+
+ @AfterEach
+ void stop() {
+ connector.stop();
+ context.stop();
+ }
+
+ private JsonObject send(JsonObject request) throws Exception {
+ AtomicReference<JsonObject> answer = new AtomicReference<>();
+ assertThat(connector.dispatchAsync(request, answer::set).get(20,
TimeUnit.SECONDS)).isTrue();
+ return answer.get();
+ }
+
+ private static JsonObject request(String body) {
+ JsonObject request = new JsonObject();
+ request.put("action", "send");
+ request.put("endpoint", "direct:refunds");
+ request.put("exchangePattern", "InOnly");
+ request.put("body", body);
+ return request;
+ }
+
+ @Test
+ void theSameMessageAgainWithAPause() throws Exception {
+ JsonObject request = request("R-1");
+ request.put("repeat", 5);
+ request.put("interval", 20);
+
+ JsonObject answer = send(request);
+
+
assertThat(answer.getString("status")).as(answer.toJson()).isEqualTo("success");
+ assertThat(answer.getLong("messages")).isEqualTo(5);
+ assertThat(answer.getLong("sent")).isEqualTo(5);
+ assertThat(answer.getLong("failed")).isZero();
+ assertThat(answer.getString("endpoint")).isEqualTo("direct://refunds");
+ // four pauses of 20 millis
+ assertThat(answer.getLong("elapsed")).isGreaterThanOrEqualTo(80);
+ assertThat(received).hasSize(5).containsOnly("R-1 -");
+ }
+
+ @Test
+ void aPauseDrawnFromARange() throws Exception {
+ JsonObject request = request("R-1");
+ request.put("repeat", 5);
+ request.put("interval", 20);
+ request.put("intervalMax", 60);
+
+ JsonObject answer = send(request);
+
+ assertThat(answer.getLong("sent")).isEqualTo(5);
+ // four pauses of 20 to 60 millis
+ assertThat(answer.getLong("elapsed")).isGreaterThanOrEqualTo(80);
+ }
+
+ @Test
+ void anIntervalIsATimeOrARange() {
+ assertThat(RepeatedSend.interval("200")).containsExactly(200, 200);
+ assertThat(RepeatedSend.interval("200ms")).containsExactly(200, 200);
+ assertThat(RepeatedSend.interval("100..500")).containsExactly(100,
500);
+ assertThat(RepeatedSend.interval("100ms..2s")).containsExactly(100,
2000);
+ assertThat(RepeatedSend.interval(null)).containsExactly(0, 0);
+ assertThatThrownBy(() ->
RepeatedSend.interval("500..100")).hasMessageContaining("from the shortest to
the longest");
+ assertThatThrownBy(() ->
RepeatedSend.interval("1s..2m")).hasMessageContaining("at most 60s");
+ assertThatThrownBy(() ->
RepeatedSend.interval("soon")).hasMessageContaining("not a time or a range");
+ assertThatThrownBy(() ->
RepeatedSend.interval("1..2..3")).hasMessageContaining("not a time or a range");
+ }
+
+ @Test
+ void theFilesOfAFolderInNameOrderWithTheirHeaders() throws Exception {
+ Path feed = Files.createDirectories(dir.resolve("refunds-fails"));
+ Files.writeString(feed.resolve("002.json"), "R-2");
+ Files.writeString(feed.resolve("001.json"), "R-1");
+ Files.writeString(feed.resolve("001.properties"), "country=ES\n");
+ Files.writeString(feed.resolve(".hidden"), "not a message");
+
+ JsonObject answer = send(request("file:" + feed));
+
+ assertThat(answer.getLong("sent")).as(answer.toJson()).isEqualTo(2);
+ assertThat(received).containsExactly("R-1 ES", "R-2 -");
+
+ // a repeat goes round the files again, and a header of the action
goes with every message
+ received.clear();
+ JsonObject request = request("file:" + feed);
+ request.put("repeat", 5);
+ JsonObject header = new JsonObject();
+ header.put("key", "country");
+ header.put("value", "DK");
+ request.put("headers", new JsonArray(List.of(header)));
+ answer = send(request);
+ assertThat(answer.getLong("sent")).isEqualTo(5);
+ assertThat(received).containsExactly("R-1 ES", "R-2 DK", "R-1 ES",
"R-2 DK", "R-1 ES");
+ }
+
+ @Test
+ void theFailuresAreCountedWithTheirErrors() throws Exception {
+ Path feed = Files.createDirectories(dir.resolve("mixed"));
+ Files.writeString(feed.resolve("1.txt"), "good");
+ Files.writeString(feed.resolve("2.txt"), "bad");
+ JsonObject request = request("file:" + feed);
+ request.put("repeat", 4);
+
+ JsonObject answer = send(request);
+
+ assertThat(answer.getString("status")).isEqualTo("failed");
+ assertThat(answer.getLong("sent")).isEqualTo(4);
+ assertThat(answer.getLong("failed")).isEqualTo(2);
+ JsonObject error = (JsonObject)
answer.getCollection("errors").iterator().next();
+
assertThat(error.getString("message")).isEqualTo("IllegalStateException: bad
refund");
+ assertThat(error.getLong("count")).isEqualTo(2);
+ }
+
+ @Test
+ void anEndpointThatIsNotThereStopsAtTheFirstMessage() throws Exception {
+ JsonObject request = request("R-1");
+ request.put("endpoint", "nosuchroute");
+ request.put("repeat", 10);
+
+ JsonObject answer = send(request);
+
+ assertThat(answer.getLong("sent")).isEqualTo(1);
+ assertThat(answer.getLong("failed")).isEqualTo(1);
+ }
+
+ @Test
+ void whatCannotBeSent() throws Exception {
+ JsonObject poll = request("R-1");
+ poll.put("repeat", 3);
+ poll.put("poll", true);
+ assertThatThrownBy(() ->
send(poll)).isInstanceOf(ExecutionException.class)
+ .hasMessageContaining("A poll does not repeat");
+
+ JsonObject many = request("R-1");
+ many.put("repeat", RepeatedSend.MAX_MESSAGES + 1);
+ assertThatThrownBy(() -> send(many)).hasMessageContaining("At most
10000 messages");
+
+ Files.createDirectories(dir.resolve("empty"));
+ assertThatThrownBy(() -> send(request("file:" +
dir.resolve("empty")))).hasMessageContaining("No messages");
+ assertThat(received).isEmpty();
+ }
+
+ @Test
+ void oneMessageIsAPlainSend() throws Exception {
+ JsonObject request = request("R-1");
+ request.put("repeat", 1);
+ assertThat(RepeatedSend.isRepeated(request)).isFalse();
+
+ JsonObject answer = send(request);
+
+ // the answer of a single send, with no summary
+ assertThat(answer.getString("status")).isEqualTo("success");
+ assertThat(answer.containsKey("sent")).isFalse();
+ assertThat(received).containsExactly("R-1 -");
+ }
+}
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json
b/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json
index e58c1560d2bc..d04b44c65d5d 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json
+++
b/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json
@@ -3,7 +3,7 @@
{ "name": "ask", "fullName": "ask", "description": "Ask a question about a
running Camel application using AI", "sourceClass":
"org.apache.camel.dsl.jbang.core.commands.Ask", "options": [ { "names":
"--api-key", "description": "API key. Also reads ANTHROPIC_API_KEY,
OPENAI_API_KEY, WATSONX_APIKEY, or LLM_API_KEY env vars", "javaType":
"java.lang.String", "type": "string" }, { "names": "--api-type", "description":
"API type: 'ollama', 'openai', 'anthropic', or 'watsonx'", "javaType": [...]
{ "name": "bind", "fullName": "bind", "description": "DEPRECATED: Bind
source and sink Kamelets as a new Camel integration", "deprecated": true,
"sourceClass": "org.apache.camel.dsl.jbang.core.commands.bind.Bind", "options":
[ { "names": "--error-handler", "description": "Add error handler
(none|log|sink:<endpoint>). Sink endpoints are expected in the format
\"[[apigroup\/]version:]kind:[namespace\/]name\", plain Camel URIs or Kamelet
name.", "javaType": "java.lang.String", "type": " [...]
{ "name": "catalog", "fullName": "catalog", "description": "List artifacts
from Camel Catalog", "sourceClass":
"org.apache.camel.dsl.jbang.core.commands.catalog.CatalogCommand", "options": [
{ "names": "-h,--help", "description": "Display the help and sub-commands",
"javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name":
"component", "fullName": "catalog component", "description": "List components
from the Camel Catalog", "sourceClass": "org.apache.camel.dsl.jbang.co [...]
- { "name": "cmd", "fullName": "cmd", "description": "Performs commands in
the running Camel integrations, such as start\/stop route, or change logging
levels.", "sourceClass":
"org.apache.camel.dsl.jbang.core.commands.action.CamelAction", "options": [ {
"names": "-h,--help", "description": "Display the help and sub-commands",
"javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name":
"browse", "fullName": "cmd browse", "description": "Browse pending messages on
endpoints [...]
+ { "name": "cmd", "fullName": "cmd", "description": "Performs commands in
the running Camel integrations, such as start\/stop route, or change logging
levels.", "sourceClass":
"org.apache.camel.dsl.jbang.core.commands.action.CamelAction", "options": [ {
"names": "-h,--help", "description": "Display the help and sub-commands",
"javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name":
"browse", "fullName": "cmd browse", "description": "Browse pending messages on
endpoints [...]
{ "name": "completion", "fullName": "completion", "description": "Generate
completion script for bash\/zsh", "sourceClass":
"org.apache.camel.dsl.jbang.core.commands.Complete", "options": [ { "names":
"-h,--help", "description": "Display the help and sub-commands", "javaType":
"boolean", "type": "boolean" } ] },
{ "name": "config", "fullName": "config", "description": "Get and set user
configuration values", "sourceClass":
"org.apache.camel.dsl.jbang.core.commands.config.ConfigCommand", "options": [ {
"names": "-h,--help", "description": "Display the help and sub-commands",
"javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name": "get",
"fullName": "config get", "description": "Display user configuration value",
"sourceClass": "org.apache.camel.dsl.jbang.core.commands.config. [...]
{ "name": "debug", "fullName": "debug", "description": "Debug local Camel
integration", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.Debug",
"options": [ { "names": "--ago", "description": "Use ago instead of yyyy-MM-dd
HH:mm:ss in timestamp.", "javaType": "boolean", "type": "boolean" }, { "names":
"--background", "description": "Run in the background", "defaultValue":
"false", "javaType": "boolean", "type": "boolean" }, { "names":
"--background-wait", "description": "How [...]
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
index 9a6341172609..e61e4aac412d 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
@@ -22,6 +22,7 @@ import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.text.SimpleDateFormat;
+import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -36,6 +37,7 @@ import java.util.stream.Stream;
import org.apache.camel.catalog.CamelCatalog;
import org.apache.camel.catalog.DefaultCamelCatalog;
+import org.apache.camel.cli.connector.RepeatedSend;
import org.apache.camel.dsl.jbang.core.commands.CamelJBangMain;
import org.apache.camel.dsl.jbang.core.commands.MavenResolverMixin;
import org.apache.camel.dsl.jbang.core.commands.Run;
@@ -60,9 +62,17 @@ import picocli.CommandLine;
description = "Send messages to endpoints", sortOptions =
false,
showDefaultValues = true,
footer = {
+ "%nWith --repeat the message is sent that many
times, with --interval between two",
+ "(a range such as 100..500 draws each pause at
random), and a summary is printed",
+ "instead of each reply. A body of file:<folder>
sends the",
+ "files of the folder in name order, one message
per file; 001.properties has the",
+ "headers of 001.json. With --repeat it goes round
the files again.",
"%nExamples:",
" camel cmd send --endpoint=kafka:myTopic
--body='Hello World'",
- " camel cmd send hello --endpoint=direct:foo
--body='Hello'" })
+ " camel cmd send hello --endpoint=direct:foo
--body='Hello'",
+ " camel cmd send --endpoint=kafka:refunds
--body=file:refund.json --repeat=50 --interval=200ms",
+ " camel cmd send --endpoint=kafka:refunds
--body=file:refund.json --repeat=50 --interval=100..500",
+ " camel cmd send --endpoint=kafka:refunds
--body=file:test/chaos/refunds-fails/" })
public class CamelSendAction extends ActionBaseCommand {
@CommandLine.Parameters(description = "To use an existing running Camel
integration for sending the message (name or pid)",
@@ -103,6 +113,21 @@ public class CamelSendAction extends ActionBaseCommand {
description = "Message header (key=value)")
List<String> headers;
+ @CommandLine.Option(names = { "--repeat" },
+ description = "Send the message this many times (going
round the files of a folder body), and"
+ + " print a summary instead of each
reply (at most 10000)")
+ int repeat;
+
+ @CommandLine.Option(names = { "--interval" },
+ description = "The pause between two messages with
--repeat or a folder body, in millis or as a"
+ + " duration such as 200ms or 1s, or a
range such as 100..500 or 100ms..2s to draw"
+ + " each pause at random (at most 60s)")
+ String interval;
+
+ // the shortest and the longest pause of --interval, in millis
+ private long intervalMin;
+ private long intervalMax;
+
@CommandLine.Option(names = { "--timeout" }, defaultValue = "20000",
converter = TimeoutConverter.class,
description = "Timeout waiting for message to be sent
(and reply message if InOut messaging),"
+ " in millis or as a duration such as
30s")
@@ -147,6 +172,9 @@ public class CamelSendAction extends ActionBaseCommand {
private MessageTableHelper tableHelper;
+ // the limit of the integration, so the CLI refuses what the integration
would
+ private static final int MAX_MESSAGES = RepeatedSend.MAX_MESSAGES;
+
private final CamelCatalog catalog = new DefaultCamelCatalog();
public CamelSendAction(CamelJBangMain main) {
@@ -166,6 +194,22 @@ public class CamelSendAction extends ActionBaseCommand {
if (timeout < 5000) {
timeout = 5000;
}
+ if (repeat < 0 || repeat > MAX_MESSAGES) {
+ throw CliException.usage("--repeat must be between 1 and " +
MAX_MESSAGES + ", was: " + repeat);
+ }
+ try {
+ long[] pause = RepeatedSend.interval(interval);
+ intervalMin = pause[0];
+ intervalMax = pause[1];
+ } catch (IllegalArgumentException e) {
+ throw CliException.usage(e.getMessage().replace("The interval",
"--interval"));
+ }
+ if (isRepeated() && poll) {
+ throw CliException.usage("A poll does not repeat: use --repeat or
a folder body to send, or --poll alone");
+ }
+ if (isRepeated() && replyFile != null) {
+ throw CliException.usage("--reply-file saves one reply: it cannot
be used with --repeat or a folder body");
+ }
if (infra != null) {
return doCallInfra(infra);
@@ -186,16 +230,25 @@ public class CamelSendAction extends ActionBaseCommand {
String mep = (reply || replyFile != null) ? "InOut" : "InOnly";
root.put("exchangePattern", mep);
if (body != null) {
- // favour using absolute path to file as the send command can be
called from another
+ // favour using absolute path to file (or folder of messages) as
the send command can be called from another
// folder than where camel is running
if (body.startsWith("file:")) {
File f = new File(body.substring(5));
- if (f.exists() && f.isFile()) {
+ if (f.exists()) {
body = "file:" + f.getAbsolutePath();
}
}
root.put("body", body);
}
+ if (repeat > 0) {
+ root.put("repeat", repeat);
+ }
+ if (intervalMax > 0) {
+ root.put("interval", intervalMin);
+ if (intervalMax > intervalMin) {
+ root.put("intervalMax", intervalMax);
+ }
+ }
if (headers != null) {
JsonArray arr = new JsonArray();
for (String h : headers) {
@@ -213,7 +266,39 @@ public class CamelSendAction extends ActionBaseCommand {
this.pid = process.pid();
// wait longer than the timeout of sending
- return showStatus(ActionClient.execute(process, "send",
this::sendOptions, timeout + 10000));
+ return showStatus(ActionClient.execute(process, "send",
this::sendOptions, waitMillis()));
+ }
+
+ /** Whether more than one message is sent: a repeat, or the files of a
folder. */
+ private boolean isRepeated() {
+ return repeat > 1 || folder() != null;
+ }
+
+ /** The folder of messages the body names (file:<folder>), when it is
here. */
+ private File folder() {
+ if (body == null || !body.startsWith("file:")) {
+ return null;
+ }
+ File f = new File(body.substring(5));
+ return f.isDirectory() ? f : null;
+ }
+
+ /**
+ * How long to wait for the answer: longer than the timeout of sending,
and for more messages the pauses between
+ * them and a second for each.
+ */
+ long waitMillis() {
+ long wait = timeout + 10000;
+ if (isRepeated()) {
+ long count = repeat;
+ File folder = folder();
+ if (count <= 1 && folder != null) {
+ String[] files = folder.list((d, n) -> !n.startsWith(".") &&
!n.endsWith(".properties"));
+ count = files != null ? files.length : 1;
+ }
+ wait += count * (intervalMax + 1000);
+ }
+ return wait;
}
/**
@@ -229,6 +314,9 @@ public class CamelSendAction extends ActionBaseCommand {
throw new CliException(result.exitCode(), "Send timeout: " +
result.error());
}
printStatusLine(jo);
+ if (jo.containsKey("sent")) {
+ return showSummary(jo);
+ }
String exchangeId = jo.getString("exchangeId");
JsonObject message = jo.getMap("message");
JsonObject cause = jo.getMap("exception");
@@ -284,6 +372,17 @@ public class CamelSendAction extends ActionBaseCommand {
return CliExitCode.OK;
}
+ /** The summary of a repeat or a folder of messages: the errors, as the
status line has the counts. */
+ private int showSummary(JsonObject jo) {
+ Collection<JsonObject> errors = jo.getCollection("errors");
+ if (errors != null) {
+ for (JsonObject error : errors) {
+ printer().println(" " + error.getLong("count") + " x " +
error.getString("message"));
+ }
+ }
+ return "success".equals(jo.getString("status")) ? CliExitCode.OK :
CliExitCode.FAILURE;
+ }
+
private Integer doCallLocal() throws Exception {
AtomicReference<KameletMain> ref = new AtomicReference<>();
Run run = new Run(this.getMain()) {
@@ -310,7 +409,7 @@ public class CamelSendAction extends ActionBaseCommand {
@Override
public void run() {
try {
- status.set(showStatus(ActionClient.execute(pid, request,
timeout + 10000)));
+ status.set(showStatus(ActionClient.execute(pid, request,
waitMillis())));
} catch (CliException e) {
failure.set(e);
} catch (Exception e) {
@@ -329,8 +428,8 @@ public class CamelSendAction extends ActionBaseCommand {
t.start();
Integer exit = run.call();
- if (!latch.await(timeout + 10000, TimeUnit.MILLISECONDS)) {
- throw CliException.timeout("Send timeout: no answer within " +
(timeout + 10000) + " millis");
+ if (!latch.await(waitMillis(), TimeUnit.MILLISECONDS)) {
+ throw CliException.timeout("Send timeout: no answer within " +
waitMillis() + " millis");
}
if (exit != null && exit != 0) {
return exit;
@@ -393,7 +492,15 @@ public class CamelSendAction extends ActionBaseCommand {
boolean reply = r.containsKey("message");
String status;
Ansi.Color c = Ansi.Color.GREEN;
- if (failed) {
+ if (r.containsKey("sent")) {
+ long sent = r.getLongOrDefault("sent", 0);
+ long count = r.getLongOrDefault("failed", 0);
+ status = "Sent " + sent + (sent == 1 ? " message" : " messages")
+ + (count > 0 ? " (" + (sent - count) + " ok, " + count +
" failed)" : " (success)");
+ if (count > 0) {
+ c = Ansi.Color.RED;
+ }
+ } else if (failed) {
status = "Failed (exception)";
c = Ansi.Color.RED;
} else if (replyFile != null) {
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
index 754fcc18856d..7d56b3725e23 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolContext.java
@@ -378,8 +378,12 @@ public class ToolContext {
}
public JsonObject sendMessage(String endpoint, String body, String
headers) {
+ return sendMessage(endpoint, body, headers, 0, null);
+ }
+
+ public JsonObject sendMessage(String endpoint, String body, String
headers, int repeat, String interval) {
requireProcess();
- return RuntimeHelper.sendMessage(pid, endpoint, body, headers);
+ return RuntimeHelper.sendMessage(pid, endpoint, body, headers, repeat,
interval);
}
public List<RuntimeHelper.ProcessInfo> discoverProcesses() {
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
index 6a9f5f96fc6b..476dbf4030cc 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
@@ -27,6 +27,7 @@ import java.util.Objects;
import java.util.stream.Collectors;
import org.apache.camel.catalog.CamelCatalog;
+import org.apache.camel.cli.connector.RepeatedSend;
import
org.apache.camel.dsl.jbang.core.commands.action.CamelRouteStructureAction;
import org.apache.camel.dsl.jbang.core.common.CliException;
import org.apache.camel.dsl.jbang.core.common.ExampleHelper;
@@ -114,6 +115,19 @@ public final class ToolRegistry {
BY_NAME.put(descriptor.name(), descriptor);
}
+ /** A whole number argument, or the default when it is not given. */
+ private static int intArg(Map<String, String> args, String name, int
defaultValue) {
+ String value = args.get(name);
+ if (value == null || value.isBlank()) {
+ return defaultValue;
+ }
+ try {
+ return Integer.parseInt(value.trim());
+ } catch (NumberFormatException e) {
+ throw new ToolExecutionException(name + " must be a whole number,
was: " + value);
+ }
+ }
+
// ---- Process discovery and selection ----
private static void registerProcessTools() {
@@ -396,12 +410,22 @@ public final class ToolRegistry {
}));
register(tool("send_message",
- "Send a test message to a Camel endpoint in the running
application.")
+ "Send a test message to a Camel endpoint in the running
application. With repeat, or a body of"
+ + " file:<folder> (one message per file,
in name order; 001.properties has the"
+ + " headers of 001.json), it sends
several messages with interval millis between"
+ + " them, and answers with a summary:
sent, failed and the errors.")
.param("endpoint", "string",
"Endpoint URI to send to (e.g., direct:myRoute,
seda:queue)", true)
- .param("body", "string", "Message body to send", false)
+ .param("body", "string", "Message body to send (file:<path>
for a file, file:<folder> for its files)",
+ false)
.param("headers", "string",
"Message headers as key=value pairs separated by
newlines", false)
+ .param("repeat", "string",
+ "How many messages to send in all (at most 10000),
going round the files of a folder", false)
+ .param("interval", "string",
+ "The pause between two messages in millis, or a range
such as 100..500 to draw each pause at"
+ + " random (at most 60000)",
+ false)
// the message runs through the routes and whatever they call,
so it can change data
.readOnly(false).destructive(true)
.core(true)
@@ -410,7 +434,15 @@ public final class ToolRegistry {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolExecutionException("endpoint is
required");
}
- return ctx.sendMessage(endpoint, args.get("body"),
args.get("headers")).toJson();
+ int repeat = intArg(args, "repeat", 0);
+ try {
+ RepeatedSend.interval(args.get("interval"));
+ } catch (IllegalArgumentException e) {
+ // an interval that is not a time or a range
+ throw new ToolExecutionException(e.getMessage());
+ }
+ return ctx.sendMessage(endpoint, args.get("body"),
args.get("headers"), repeat,
+ args.get("interval")).toJson();
}));
register(tool("eval_expression",
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
index 9e6e724c50ea..97056c9d448e 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelper.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.dsl.jbang.core.common;
+import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -23,6 +24,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
+import org.apache.camel.cli.connector.RepeatedSend;
import org.apache.camel.dsl.jbang.run.common.PathUtils;
import org.apache.camel.util.json.JsonArray;
import org.apache.camel.util.json.JsonObject;
@@ -223,7 +225,22 @@ public final class RuntimeHelper {
* @throws IllegalArgumentException when a header is not a key=value pair
*/
public static JsonObject sendMessage(long pid, String endpoint, String
body, String headers) {
+ return sendMessage(pid, endpoint, body, headers, 0, null);
+ }
+
+ /**
+ * Sends a message, or with a repeat or a body of {@code file:<folder>}
several messages, with a pause between them;
+ * the answer is then a summary (sent, failed and the errors) rather than
the reply.
+ *
+ * @param repeat how many messages to send in all, going round the files
of a folder; 0 to send once (each file of
+ * a folder once)
+ * @param interval the pause between two messages: millis, a duration such
as 200ms, or a range such as 100..500 to
+ * draw each pause at random
+ */
+ public static JsonObject sendMessage(
+ long pid, String endpoint, String body, String headers, int
repeat, String interval) {
JsonArray headerList = headers != null ? parseHeaders(headers) : null;
+ long[] pause = RepeatedSend.interval(interval);
String result = executeAction(pid, "send", root -> {
root.put("endpoint", endpoint);
if (body != null) {
@@ -232,7 +249,16 @@ public final class RuntimeHelper {
if (headerList != null) {
root.put("headers", headerList);
}
- });
+ if (repeat > 0) {
+ root.put("repeat", repeat);
+ }
+ if (pause[1] > 0) {
+ root.put("interval", pause[0]);
+ if (pause[1] > pause[0]) {
+ root.put("intervalMax", pause[1]);
+ }
+ }
+ }, sendTimeout(body, repeat, pause[1]));
try {
return (JsonObject) Jsoner.deserialize(result);
} catch (Exception e) {
@@ -242,6 +268,21 @@ public final class RuntimeHelper {
}
}
+ /**
+ * How long to wait for the answer of a send: for more messages also the
pauses between them and a second for each.
+ * The files of a folder are counted when the folder is here (the
integration reads it, relative to its working
+ * directory), else 100 are assumed.
+ */
+ static long sendTimeout(String body, int repeat, long interval) {
+ long count = repeat;
+ if (count <= 1 && body != null && body.startsWith("file:")) {
+ File folder = new File(body.substring(5));
+ String[] files = folder.list((d, n) -> !n.startsWith(".") &&
!n.endsWith(".properties"));
+ count = files != null ? files.length : folder.isFile() ? 1 : 100;
+ }
+ return count <= 1 ? ACTION_TIMEOUT_MS : ACTION_TIMEOUT_MS + count *
(interval + 1000);
+ }
+
/**
* The headers as the send action of the CLI connector reads them: a key
and a value for each header.
*/
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
index d21b7a4dcf9e..3e2ca538658e 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.dsl.jbang.core.commands.action;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.List;
import java.util.Map;
import org.apache.camel.dsl.jbang.core.commands.CamelJBangMain;
@@ -125,6 +126,78 @@ class CamelSendActionResultTest extends
ActionCommandTestSupport {
assertThat(printer.getStderr()).contains("No endpoint could be found
for: foo:bar");
}
+ @Test
+ void repeatSendsASummary() throws Exception {
+ writeStatusFile(TEST_PID, "orders");
+ Path feed = Files.createDirectories(dir.resolve("refunds-fails"));
+ Files.writeString(feed.resolve("001.json"), "{}");
+ Files.writeString(feed.resolve("001.properties"), "country=ES\n");
+ Files.writeString(feed.resolve("002.json"), "{}");
+ JsonObject summary = new JsonObject();
+ summary.put("timestamp", 1_700_000_000_000L);
+ summary.put("status", "failed");
+ summary.put("elapsed", 1200L);
+ summary.put("endpoint", "kafka://refunds");
+ summary.put("messages", 50L);
+ summary.put("sent", 50L);
+ summary.put("failed", 2L);
+ JsonObject error = new JsonObject();
+ error.put("message", "ChaosException: Chaos error of rule r1");
+ error.put("count", 2L);
+ summary.put("errors", new JsonArray(List.of(error)));
+
+ CamelSendAction command = command("orders",
"--endpoint=kafka:refunds", "--body=file:" + feed, "--repeat=50",
+ "--interval=200ms");
+ int exit = callWithResponse(command, summary);
+
+ assertThat(exit).isEqualTo(CliExitCode.FAILURE);
+ JsonObject action = readActionFile(TEST_PID);
+ assertThat(action.getString("body")).isEqualTo("file:" +
feed.toAbsolutePath());
+ assertThat(action.getLong("repeat")).isEqualTo(50L);
+ assertThat(action.getLong("interval")).isEqualTo(200L);
+ assertThat(printer.getStdout()).contains("kafka://refunds", "Sent 50
messages (48 ok, 2 failed)")
+ .contains("2 x ChaosException: Chaos error of rule r1");
+ // the wait: the timeout, and the pauses and a second for each of the
50 messages
+ assertThat(command.waitMillis()).isEqualTo(20000 + 10000 + 50 * 1200);
+
+ // a folder without a repeat: one message per file, which the wait
counts
+ CamelSendAction folder = command("orders", "--endpoint=kafka:refunds",
"--body=file:" + feed);
+ assertThat(folder.waitMillis()).isEqualTo(20000 + 10000 + 2 * 1000);
+ }
+
+ @Test
+ void whatCannotBeRepeated() {
+ assertThatThrownBy(() -> command("orders", "--endpoint=direct:start",
"--repeat=3", "--poll").doCall())
+ .isInstanceOf(CliException.class).hasMessageContaining("A poll
does not repeat")
+ .extracting(e -> ((CliException)
e).getExitCode()).isEqualTo(CliExitCode.USAGE);
+ assertThatThrownBy(() -> command("orders", "--endpoint=direct:start",
"--repeat=3", "--reply-file=x.json")
+ .doCall()).hasMessageContaining("--reply-file saves one
reply");
+ assertThatThrownBy(() -> command("orders", "--endpoint=direct:start",
"--repeat=10001").doCall())
+ .hasMessageContaining("--repeat must be between 1 and 10000");
+ assertThatThrownBy(() -> command("orders", "--endpoint=direct:start",
"--repeat=3", "--interval=2m").doCall())
+ .hasMessageContaining("--interval must be at most 60s");
+ assertThatThrownBy(() -> command("orders", "--endpoint=direct:start",
"--repeat=3", "--interval=500..100").doCall())
+ .isInstanceOf(CliException.class).hasMessageContaining("from
the shortest to the longest pause");
+ }
+
+ @Test
+ void aRangeDrawsEachPauseAtRandom() throws Exception {
+ writeStatusFile(TEST_PID, "orders");
+ JsonObject summary = new JsonObject(
+ Map.of("status", "success", "sent", 10L, "failed", 0L,
"elapsed", 3000L,
+ "endpoint", "direct://start", "errors", new
JsonArray()));
+
+ CamelSendAction command = command("orders", "--endpoint=direct:start",
"--repeat=10", "--interval=100..500");
+ assertThat(callWithResponse(command,
summary)).isEqualTo(CliExitCode.OK);
+
+ JsonObject action = readActionFile(TEST_PID);
+ assertThat(action.getLong("interval")).isEqualTo(100L);
+ assertThat(action.getLong("intervalMax")).isEqualTo(500L);
+ // the wait counts the longest pause
+ assertThat(command.waitMillis()).isEqualTo(20000 + 10000 + 10 * 1500);
+ assertThat(printer.getStdout()).contains("Sent 10 messages (success)");
+ }
+
@Test
void headerWithoutValueIsAUsageError() {
assertThatThrownBy(() -> command("orders", "--endpoint=direct:start",
"--header=id").doCall())
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
index 0cc151139ba5..faa59ecfc6a9 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
@@ -214,6 +214,17 @@ class RuntimeToolsTest {
.hasMessage("Header must be in key=value format, was: id");
}
+ @Test
+ void anIntervalThatIsNotATimeOrARangeIsAToolError() {
+ answers.put("send", Map.of("status", "success"));
+
+ assertThatThrownBy(() -> call("send_message", Map.of("endpoint",
"direct:orders", "repeat", "5",
+ "interval", "500..100")))
+ .isInstanceOf(ToolExecutionException.class)
+ .hasMessageContaining("from the shortest to the longest");
+ assertThat(integration.lastAction()).isNull();
+ }
+
@Test
void aSqlQuery() throws Exception {
answers.put("sql-query", Map.of("columns", List.of("id"), "rows",
List.of()));
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelperTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelperTest.java
index 967152e936cd..02c9481b5b1f 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelperTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/RuntimeHelperTest.java
@@ -142,4 +142,21 @@ class RuntimeHelperTest {
assertEquals(0, files.count());
}
}
+
+ @Test
+ void sendTimeoutCountsTheMessages() throws IOException {
+ // one message: the time of an action
+ assertEquals(10_000, RuntimeHelper.sendTimeout("hello", 0, 0));
+ assertEquals(10_000, RuntimeHelper.sendTimeout("hello", 1, 500));
+ // a repeat: the pauses between the messages, and a second for each
+ assertEquals(10_000 + 50 * 1200, RuntimeHelper.sendTimeout("hello",
50, 200));
+ // the files of a folder that is here, leaving out the headers of a
message
+ Path feed = Files.createDirectories(home.resolve("feed"));
+ Files.writeString(feed.resolve("001.json"), "{}");
+ Files.writeString(feed.resolve("001.properties"), "country=ES");
+ Files.writeString(feed.resolve("002.json"), "{}");
+ assertEquals(10_000 + 2 * 1000, RuntimeHelper.sendTimeout("file:" +
feed, 0, 0));
+ // a folder of the integration that is not here: 100 messages assumed
+ assertEquals(10_000 + 100 * 1000, RuntimeHelper.sendTimeout("file:" +
feed.resolve("elsewhere"), 0, 0));
+ }
}
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt
b/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt
index 9f7445566d67..dd3453c3640e 100644
--- a/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt
+++ b/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt
@@ -1356,9 +1356,9 @@ Usage: camel cmd send [-h] [--fresh] [--poll] [--pretty]
[--reply]
[=<loggingColor>]] [--show-body[=<showBody>]]
[--show-exception[=<showException>]] [--show-headers
[=<showHeaders>]] [--body=<body>] [--infra=<infra>]
- [--properties=<propertiesFiles>]
- [--reply-file=<replyFile>] [--repo=<repos>]
- [--timeout=<timeout>] [--uri=<endpoint>]
+ [--interval=<interval>] [--properties=<propertiesFiles>]
+ [--repeat=<repeat>] [--reply-file=<replyFile>]
+ [--repo=<repos>] [--timeout=<timeout>] [--uri=<endpoint>]
[--header=<headers>]... [--prop=<property>]... [<name>]
Send messages to endpoints
[<name>] To use an existing running Camel integration for
@@ -1387,6 +1387,15 @@ Send messages to endpoints
--body=<body> Message body to send (prefix with file: to refer to
loading message body from file)
--header=<headers> Message header (key=value)
+ --repeat=<repeat> Send the message this many times (going round the
+ files of a folder body), and print a summary
+ instead of each reply (at most 10000)
+ Default: 0
+ --interval=<interval> The pause between two messages with --repeat or a
+ folder body, in millis or as a duration such as
+ 200ms or 1s, or a range such as 100..500 or
+ 100ms..2s to draw each pause at random (at most
+ 60s)
--timeout=<timeout> Timeout waiting for message to be sent (and reply
message if InOut messaging), in millis or as a
duration such as 30s
@@ -1417,9 +1426,20 @@ Send messages to endpoints
repositories)
--fresh Make sure we use fresh (i.e. non-cached) resources
+With --repeat the message is sent that many times, with --interval between two
+(a range such as 100..500 draws each pause at random), and a summary is printed
+instead of each reply. A body of file:<folder> sends the
+files of the folder in name order, one message per file; 001.properties has the
+headers of 001.json. With --repeat it goes round the files again.
+
Examples:
camel cmd send --endpoint=kafka:myTopic --body='Hello World'
camel cmd send hello --endpoint=direct:foo --body='Hello'
+ camel cmd send --endpoint=kafka:refunds --body=file:refund.json --repeat=50
+--interval=200ms
+ camel cmd send --endpoint=kafka:refunds --body=file:refund.json --repeat=50
+--interval=100..500
+ camel cmd send --endpoint=kafka:refunds --body=file:test/chaos/refunds-fails/
== camel cmd span --help
Usage: camel cmd span [-h] [--flat] [--json] [--logging-color[=<loggingColor>]]
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
index d15661829100..1a13c17f93ed 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
@@ -533,13 +533,23 @@ public class RuntimeTools {
@ToolGuardrails(input = McpArgumentGuardrail.class, output =
McpArgumentGuardrail.class)
@Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = true, openWorldHint = false),
- description = "Send a test message to a Camel endpoint in the
running application.")
+ description = """
+ Send a test message to a Camel endpoint in the running
application. With repeat, or a body of \
+ file:<folder> (one message per file, in name order;
001.properties has the headers of 001.json), it \
+ sends several messages with interval millis between them,
and answers with a summary: sent, failed \
+ and the errors. A broker consumer stubbed with camel run
--stub=remote is fed this way.""")
public JsonObject camel_runtime_send(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
@ToolArg(description = "Endpoint URI to send to (e.g.,
direct:myRoute, seda:queue)") String endpoint,
- @ToolArg(description = "Message body to send", required = false)
String body,
+ @ToolArg(description = "Message body to send (file:<path> for a
file, file:<folder> for its files)",
+ required = false) String body,
@ToolArg(description = "Message headers as key=value pairs
separated by newlines",
- required = false) String headers) {
+ required = false) String headers,
+ @ToolArg(description = "How many messages to send in all (at most
10000), going round the files of a folder",
+ required = false) Integer repeat,
+ @ToolArg(description = "The pause between two messages in millis,
or a range such as 100..500 to draw"
+ + " each pause at random (at most 60000)",
+ required = false) String interval) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
@@ -551,6 +561,12 @@ public class RuntimeTools {
if (headers != null) {
args.put("headers", headers);
}
+ if (repeat != null) {
+ args.put("repeat", String.valueOf(repeat));
+ }
+ if (interval != null) {
+ args.put("interval", interval);
+ }
return delegateToRegistry("send_message", nameOrPid, args);
}
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
index 4cdeac541d40..af40f151361b 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
@@ -134,7 +134,7 @@ class RuntimeToolsTest {
@Test
void sendRequiresEndpoint() {
RuntimeTools tools = createTools();
- assertThatThrownBy(() -> tools.camel_runtime_send(null, null, "body",
null))
+ assertThatThrownBy(() -> tools.camel_runtime_send(null, null, "body",
null, null, null))
.isInstanceOf(ToolCallException.class)
.hasMessageContaining("endpoint is required");
}
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
index d664bb379cec..76b3d3c9a471 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
@@ -396,7 +396,8 @@ class RuntimeToolsWiringTest {
void theHintsMatchWhatTheToolsDo() throws Exception {
assertThat(hints("camel_runtime_receive", String.class, String.class,
Integer.class, Integer.class))
.as("receive takes the message from the
routes").containsExactly(false, true);
- assertThat(hints("camel_runtime_send", String.class, String.class,
String.class, String.class))
+ assertThat(hints("camel_runtime_send", String.class, String.class,
String.class, String.class, Integer.class,
+ String.class))
.containsExactly(false, true);
assertThat(hints("camel_runtime_trace_dump",
String.class)).containsExactly(true, false);
