This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26795-chaos-file-feed in repository https://gitbox.apache.org/repos/asf/camel.git
commit 37b141ea46e3240294239abccf0c5c280ee47e60 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 9 23:30:44 2026 +0100 CAMEL-26795: camel-jbang - a chaos experiment feeds a file route with files its filter takes, and camel cmd send --repeat numbers a CamelFileName - a feed to a route that starts from file:<folder> names its files so the route takes them: chaos-<time>.<ext>, the extension of its include or antInclude filter, else of the body; without it a route with include=.*\.json skipped the files, and the experiment failed as if the route lost them - a repeated send (or a folder) numbers a CamelFileName header per message, so a file endpoint writes one file per message: order-1.json, order-2.json Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../modules/ROOT/pages/camel-jbang-devtools.adoc | 4 ++ .../modules/ROOT/pages/camel-jbang-managing.adoc | 4 ++ .../apache/camel/cli/connector/RepeatedSend.java | 20 ++++++ .../connector/LocalCliConnectorSendActionTest.java | 24 +++++++ .../core/commands/ai/ChaosExplorePlaybook.java | 4 +- .../dsl/jbang/core/common/ChaosExperimentRun.java | 78 +++++++++++++++++++++- .../jbang/core/common/ChaosExperimentRunTest.java | 38 +++++++++++ 7 files changed, 169 insertions(+), 3 deletions(-) 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 6ca1d495ee63..019ef6c84562 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-jbang-devtools.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-devtools.adoc @@ -73,6 +73,10 @@ $ camel cmd send shop --endpoint=kafka:refunds --body=file:refund.json --repeat= 2 x IllegalStateException: Refund amount is missing ---- +When the messages are repeated (or come from a folder), a `CamelFileName` header is numbered per message, so a file +endpoint writes one file per message rather than the same file again: `--header=CamelFileName=order.json` writes +`order-1.json`, `order-2.json` and so on. + 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: diff --git a/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc b/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc index 05efca7c4064..d81888542448 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc @@ -967,6 +967,10 @@ camel cmd chaos-experiment --hypothesis='a client of payment gets an error, noth For a path with a parameter, `--feed` is the path to call, with an optional method: `--feed='PUT /orders/42'`. An HTTP server with its own address (netty-http, undertow, jetty) cannot be fed this way yet. +A route that starts from `file:<folder>` is fed with files: each message is written as a file into the folder the route +polls, named so the route takes it (`chaos-<time>-<n>.json`, with the extension of its `include` or `antInclude` filter, +else of the body). A `CamelFileName` in `--header` names the files instead. + A feed to a remote system that is not stubbed, such as a Kafka broker, has a warning, as the messages go to that system: use a local one (`camel infra run kafka`), or stub it. 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 index a1956dd53f6d..afe6b54c53a4 100644 --- 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 @@ -51,6 +51,9 @@ public final class RepeatedSend { 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; + + // the name of the file a file endpoint writes a message to + private static final String FILE_NAME = "CamelFileName"; private static final int MAX_ERRORS = 5; private RepeatedSend() { @@ -92,6 +95,19 @@ public final class RepeatedSend { return answer; } + /** + * A file name with the number of the message before its extension: order.json is order-3.json for the third + * message, and order is order-3. + */ + static String numbered(String name, long number) { + int slash = Math.max(name.lastIndexOf('/'), name.lastIndexOf('\\')); + int dot = name.lastIndexOf('.'); + if (dot <= slash + 1) { + return name + "-" + number; + } + return name.substring(0, dot) + "-" + number + name.substring(dot); + } + /** 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; @@ -191,6 +207,10 @@ public final class RepeatedSend { if (message.body() != null) { args.put("body", message.body()); } + // a file endpoint gets one file per message, rather than the same file again (CAMEL-26795) + if (count > 1 && args.get(FILE_NAME) instanceof String name && !name.isBlank()) { + args.put(FILE_NAME, numbered(name, i + 1)); + } JsonObject response = (JsonObject) console.call(DevConsole.MediaType.JSON, args); if (endpoint == null) { endpoint = response.getString("endpoint"); 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 index 3bdb264a73c8..d3f24c3b38d9 100644 --- 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 @@ -126,6 +126,30 @@ class LocalCliConnectorSendActionTest { assertThat(answer.getLong("elapsed")).isGreaterThanOrEqualTo(80); } + @Test + void aRepeatedFileNameIsNumbered() throws Exception { + // a file endpoint writes one file per message (CAMEL-26795) + Path inbox = dir.resolve("inbox"); + JsonObject request = request("{}"); + request.put("endpoint", "file:" + inbox); + request.put("repeat", 3); + JsonObject header = new JsonObject(); + header.put("key", "CamelFileName"); + header.put("value", "order.json"); + request.put("headers", new JsonArray(List.of(header))); + + JsonObject answer = send(request); + + assertThat(answer.getLong("sent")).as(answer.toJson()).isEqualTo(3); + try (var files = Files.list(inbox)) { + assertThat(files.map(f -> f.getFileName().toString()).sorted()) + .containsExactly("order-1.json", "order-2.json", "order-3.json"); + } + assertThat(RepeatedSend.numbered("order", 2)).isEqualTo("order-2"); + assertThat(RepeatedSend.numbered("in.box/order", 2)).isEqualTo("in.box/order-2"); + assertThat(RepeatedSend.numbered(".hidden", 2)).isEqualTo(".hidden-2"); + } + @Test void anIntervalIsATimeOrARange() { assertThat(RepeatedSend.interval("200")).containsExactly(200, 200); diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ChaosExplorePlaybook.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ChaosExplorePlaybook.java index f51340170df7..3e5d9b951224 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ChaosExplorePlaybook.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ChaosExplorePlaybook.java @@ -91,8 +91,8 @@ public final class ChaosExplorePlaybook { only when the route feeds itself (kind self). A route of kind http (platform-http or rest) is fed \ with HTTP requests to the server of the integration: give the id of the route as feed, or the path \ to call when its path has a parameter (such as POST /orders/42); the traffic counts the status codes \ - the clients got. Routes fed from files cannot be fed by an experiment yet. An experiment refuses a \ - feed that cannot reach its route, with the reason and \ + the clients got. A route of kind file is fed with its id too: each message is written as a file the \ + route takes. An experiment refuses a feed that cannot reach its route, with the reason and \ what to do instead, and warns when the feed goes to a real system that is not stubbed - a rule at=dlq fails the dead letter channel, so it fires only when a message goes there: use \ the combined suggestions, which add the fault that sends messages there too. A message the \ diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java index 6e38b8230f4a..0e8757c4effd 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java @@ -16,8 +16,11 @@ */ package org.apache.camel.dsl.jbang.core.common; +import java.io.File; import java.io.IOException; import java.net.URI; +import java.net.URISyntaxException; +import java.net.URLDecoder; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; @@ -25,6 +28,7 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.StringJoiner; import java.util.function.Consumer; @@ -35,6 +39,7 @@ import org.apache.camel.catalog.CamelCatalog; import org.apache.camel.catalog.DefaultCamelCatalog; import org.apache.camel.tooling.model.ComponentModel; import org.apache.camel.util.TimeUtils; +import org.apache.camel.util.URISupport; import org.apache.camel.util.json.JsonObject; /** @@ -179,7 +184,17 @@ public final class ChaosExperimentRun { } // a route that serves HTTP is fed with requests to the server of the integration List<Object> routes = status != null ? status.getCollection("routes") : null; - http = ChaosHttpFeed.target(status, routes != null ? route(routes, feed.endpoint()) : null, feed.endpoint()); + JsonObject route = routes != null && !feed.endpoint().endsWith("*") ? route(routes, feed.endpoint()) : null; + http = ChaosHttpFeed.target(status, route, feed.endpoint()); + // a file route reads the files its filter takes: the messages are written as such files + String from = route != null ? route.getString("from") : null; + String name = from != null && from.startsWith("file:") ? fileName(from, feed, clock.now()) : null; + if (name != null) { + feed = new Feed( + feed.endpoint(), feed.body(), + (feed.headers() != null ? feed.headers() + "\n" : "") + "CamelFileName=" + name, feed.repeat(), + feed.interval()); + } } // an expectation such as failed<=0 is decided when its within= is up: a shorter wait leaves it undecided long within = longestWithin(rules); @@ -258,6 +273,67 @@ public final class ChaosExperimentRun { return answer; } + // the extension in an include filter (.*\.json, .*\.(csv|txt)) or an antInclude one (*.json) + private static final Pattern INCLUDE_EXT = Pattern.compile("\\\\\\.\\(?([A-Za-z0-9]+)"); + private static final Pattern ANT_EXT = Pattern.compile("\\*\\.([A-Za-z0-9]+)"); + + /** + * The name of the files of a feed to a route that starts from file:<folder> (CAMEL-26795), so the route takes them: + * chaos-<time>.<extension>, the extension from its include or antInclude filter, else from the body. The send + * numbers it per message when it repeats. Null when the feed names its files itself, or the route reads one file + * (fileName=), which the send writes as the endpoint has it. + */ + static String fileName(String from, Feed feed, long now) { + if (feed.headers() != null && feed.headers().toLowerCase(Locale.ROOT).contains("camelfilename")) { + return null; + } + Map<String, Object> options; + try { + options = URISupport.parseQuery(URISupport.extractQuery(from)); + } catch (URISyntaxException e) { + options = Map.of(); + } + if (options.containsKey("fileName")) { + return null; + } + String ext = null; + if (options.get("include") instanceof String include) { + Matcher m = INCLUDE_EXT.matcher(decode(include)); + ext = m.find() ? m.group(1) : null; + } + if (ext == null && options.get("antInclude") instanceof String ant) { + Matcher m = ANT_EXT.matcher(decode(ant)); + ext = m.find() ? m.group(1) : null; + } + if (ext == null) { + ext = extension(feed.body()); + } + return "chaos-" + now + "." + ext; + } + + // an option of the endpoint URI as the status has it, %5C for a backslash; a + stays a +, as in a regular expression + private static String decode(String value) { + try { + return URLDecoder.decode(value.replace("+", "%2B"), StandardCharsets.UTF_8); + } catch (IllegalArgumentException e) { + return value; + } + } + + // the extension of a body: of the file it names, else json, xml or txt by its first character + private static String extension(String body) { + if (body == null) { + return "txt"; + } + if (body.startsWith("file:")) { + String name = new File(body.substring(5)).getName(); + int dot = name.lastIndexOf('.'); + return dot > 0 && dot < name.length() - 1 ? name.substring(dot + 1) : "txt"; + } + String b = body.stripLeading(); + return b.startsWith("{") || b.startsWith("[") ? "json" : b.startsWith("<") ? "xml" : "txt"; + } + /** The longest within= of the rules, in millis, or 0 (an expectation without it has the time of its rule). */ static long longestWithin(String rules) { long answer = 0; diff --git a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java index d731f011a069..1b8c9db911c2 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java +++ b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java @@ -360,6 +360,44 @@ class ChaosExperimentRunTest { .hasMessageContaining("the port of the HTTP server of the integration is not known"); } + @Test + void theFilesOfAFileFeedPassTheFilterOfTheRoute() { + ChaosExperimentRun.Feed json = new ChaosExperimentRun.Feed("invoices", "{\"n\": 1}", null, 5, null); + // the extension of the include or antInclude filter (CAMEL-26795) + assertThat(ChaosExperimentRun.fileName("file://invoices?delete=true&include=.*%5C.json", json, 42)) + .isEqualTo("chaos-42.json"); + assertThat(ChaosExperimentRun.fileName("file://in?include=%28%3Fi%29.*%5C.%28csv%7Ctxt%29", json, 42)) + .isEqualTo("chaos-42.csv"); + assertThat(ChaosExperimentRun.fileName("file://in?antInclude=*.xml", json, 42)).isEqualTo("chaos-42.xml"); + assertThat(ChaosExperimentRun.fileName("file://invoices?include=.*%5C.json", + new ChaosExperimentRun.Feed("invoices", "plain text", null, 0, null), 42)).isEqualTo("chaos-42.json"); + // no filter: by the body, or the file it names + assertThat(ChaosExperimentRun.fileName("file://in", json, 42)).isEqualTo("chaos-42.json"); + assertThat(ChaosExperimentRun.fileName("file://in", + new ChaosExperimentRun.Feed("in", "<order/>", null, 0, null), 42)).isEqualTo("chaos-42.xml"); + assertThat(ChaosExperimentRun.fileName("file://in", + new ChaosExperimentRun.Feed("in", "file:/feed/order.csv", null, 0, null), 42)).isEqualTo("chaos-42.csv"); + // the feed names its files, or the route reads one file + assertThat(ChaosExperimentRun.fileName("file://in", + new ChaosExperimentRun.Feed("in", "x", "CamelFileName=mine.txt", 0, null), 42)).isNull(); + assertThat(ChaosExperimentRun.fileName("file://in?fileName=orders.csv", json, 42)).isNull(); + } + + @Test + void aFileRouteIsFedWithNamedFiles() throws Exception { + JsonObject status = statusWith(experiment("PENDING", true)); + status.put("routes", new JsonArray( + List.of( + new JsonObject(Map.of("routeId", "invoices", "from", "file://invoices?include=.*%5C.json"))))); + integration.statuses.add(status); + integration.shows("FAIL", true); + + ChaosExperimentRun.run(integration, null, RULE, + new ChaosExperimentRun.Feed("invoices", "{}", "country=DK", 5, null), 60_000, null, clock); + + assertThat(integration.sent.headers()).isEqualTo("country=DK\nCamelFileName=chaos-1000000.json"); + } + @Test void aFeedToARealSystemIsWarned() { JsonObject report = new JsonObject(
