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 c8c7b9dd67f1 CAMEL-26719: camel-chaos - when= fires a rule only for
the messages that meet a condition (#29389)
c8c7b9dd67f1 is described below
commit c8c7b9dd67f1cfc21ce469c228c39116e0921a1e
Author: Claus Ibsen <[email protected]>
AuthorDate: Sat Nov 7 22:01:24 2026 +0100
CAMEL-26719: camel-chaos - when= fires a rule only for the messages that
meet a condition (#29389)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/chaos-model.json | 13 ++
.../org/apache/camel/catalog/docs/chaos.adoc | 35 +++-
.../org/apache/camel/chaos/chaos-model.json | 13 ++
components/camel-chaos/src/main/docs/chaos.adoc | 35 +++-
.../apache/camel/component/chaos/ChaosManager.java | 31 +++-
.../apache/camel/component/chaos/ChaosModel.java | 11 ++
.../component/chaos/ChaosRoutePolicyFactory.java | 3 +-
.../apache/camel/component/chaos/ChaosRule.java | 64 ++++++++
.../camel/component/chaos/ChaosRuleParser.java | 84 ++++++++--
.../camel/component/chaos/ChaosSendHook.java | 6 +-
.../camel/component/chaos/ChaosStepHook.java | 3 +-
.../camel/component/chaos/ChaosMainTest.java | 11 ++
.../camel/component/chaos/ChaosModelTest.java | 3 +-
.../component/chaos/ChaosRuleDescribeTest.java | 2 +
.../camel/component/chaos/ChaosRuleParserTest.java | 37 ++++-
.../chaos/ChaosWhenNoStreamCachingTest.java | 57 +++++++
.../camel/component/chaos/ChaosWhenTest.java | 180 +++++++++++++++++++++
.../core/commands/action/CamelChaosAddAction.java | 1 +
.../src/test/resources/golden/cli-help.txt | 2 +
.../dsl/jbang/core/commands/tui/ChaosInfo.java | 4 +-
.../core/commands/tui/ChaosTabRenderTest.java | 4 +
21 files changed, 575 insertions(+), 24 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json
index 6c3a7a52f5e2..214f24a833aa 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json
@@ -195,6 +195,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -231,6 +232,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -256,6 +258,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -303,6 +306,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -326,6 +330,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -481,6 +486,14 @@
"description": "When the rule ends: a number of matches or a time from
its start (at most camel.chaos.maxDuration, its default)",
"example": "10m"
},
+ {
+ "name": "when",
+ "type": "predicate",
+ "required": false,
+ "description": "A condition on the message in the simple language, in
quotes: the rule fires only for the messages that meet it, and its rate and
until count only those. Not with at=poll, stall or lifecycle; a condition on
the body needs stream caching",
+ "example": "\"${header.country} == 'ES'\"",
+ "language": "simple"
+ },
{
"name": "within",
"type": "duration",
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
index 54f0b101cc80..3ead495e3743 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
@@ -171,8 +171,8 @@ only the list and the report.
== Rules
-A rule is one line: the target, the fault and its parameters as `name=value`,
then `at`, `in`, `rate`, `from` and
-`until`, and optionally an expectation. A value with spaces or a `;` is quoted
with `"` or `'` (`\"`, `\'` and `\\` inside quotes
+A rule is one line: the target, the fault and its parameters as `name=value`,
then `at`, `in`, `when`, `rate`,
+`from` and `until`, and optionally an expectation. A value with spaces or a
`;` is quoted with `"` or `'` (`\"`, `\'` and `\\` inside quotes
are a quote and a backslash).
[source]
@@ -182,6 +182,8 @@ fault: error | delay | drop | corrupt | duplicate | stall
| lifecycle, with it
at: send | reply | dlq | poll | entry | step (optional)
in: in=<route id> narrows the sends, polls and steps of endpoint=,
component= and step= to a route
(* matches any characters; optional)
+when: a condition on the message in the simple language, in quotes
(when="${header.country} == 'ES'");
+ the rule fires only for the messages that meet it (optional)
rate: a probability (rate=0.3), every Nth match (rate=every:5), or bursts
(rate=burst:5s/30s: every match
during the first 5s of each 30s); without it every match fires
from: when the rule starts after it is added (from=+30s), such as after the
warm-up; without it at once
@@ -227,6 +229,32 @@ rule is about its route already, and `stall` and
`lifecycle` act on the route of
When no route matches `in=`, or the target matches no send, poll or step
inside those routes, adding the rule warns,
with what is there.
+=== A condition on the message
+
+`when=` fires a rule only for the messages that meet a condition, written in
the
+xref:languages:simple-language.adoc[Simple] language, in quotes:
+
+[source]
+----
+endpoint=http:stock* error status=503 when="${header.country} == 'ES'"
+route=orders error rate=0.1 when="${header.priority} == 'high'"
+endpoint=kafka:orders* duplicate when="${body} contains 'refund'"
+----
+
+The condition is made into a predicate once, when the rule is added (a
condition that is not valid simple is refused,
+with the reason of the simple language), and checked before the rate: `rate`
and an `until` count count only the
+messages that meet it, and the seed gives the same faults for the same
matching messages. "Fail 5% of the orders, or
+the ones from Spain" is two rules, one with `rate=0.05` and one with `when=`.
A condition that fails, such as a method
+call on a header that is not there, is not met: the exchange goes on, and the
report counts it in `skippedReasons`.
+
+A poll has no message, so a rule with `when=` leaves out the polls of its
target, and `at=poll`, `stall` and
+`lifecycle` take no `when=`. A condition that reads the body (`+${body}+`,
`+${bodyAs(...)}+`, `jsonpath`, `jq`, `xpath`)
+reads a stream body from the stream cache, and resets the cache after the
check, so the route reads the whole body; a
+reply that is a stream (`at=reply`) is cached first. Without stream caching a
stream would be consumed, so such a
+condition is refused when stream caching is off. Only simple is supported for
now (a language in front, such as
+`groovy:`, is refused). On the command line, quote the rule with `'` and a
text in the condition with `\"`:
+`+--rule='endpoint=http:stock* error when="${header.country} == \"ES\""'+`.
+
A rule gets an id when it is added (`r1`, `r2`, ...), which change and remove
take. A change replaces the rule with a
new one with the same id and the next version (a rule is version 1 when it is
added), and the replaced one stays in the
report: so the report has an entry per version of the id, and a tool tells
them apart by the `version` of the entry
@@ -646,6 +674,9 @@ seconds
|`route=orders error rate=0.05`
|fail 5% of the messages entering route orders
+|`+endpoint=http:stock* error status=503 when="${header.country} == 'ES'"+`
+|+fail every send to http:stock* with HTTP status 503, when ${header.country}
== 'ES'+
+
|`route=poller lifecycle action=remove duration=30s repeat=3 every=1m`
|remove route poller for 30 seconds, then add it back, 3 times, every 1 minute
|===
diff --git
a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json
b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json
index 6c3a7a52f5e2..214f24a833aa 100644
---
a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json
+++
b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json
@@ -195,6 +195,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -231,6 +232,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -256,6 +258,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -303,6 +306,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -326,6 +330,7 @@
"in",
"rate",
"until",
+ "when",
"within"
]
},
@@ -481,6 +486,14 @@
"description": "When the rule ends: a number of matches or a time from
its start (at most camel.chaos.maxDuration, its default)",
"example": "10m"
},
+ {
+ "name": "when",
+ "type": "predicate",
+ "required": false,
+ "description": "A condition on the message in the simple language, in
quotes: the rule fires only for the messages that meet it, and its rate and
until count only those. Not with at=poll, stall or lifecycle; a condition on
the body needs stream caching",
+ "example": "\"${header.country} == 'ES'\"",
+ "language": "simple"
+ },
{
"name": "within",
"type": "duration",
diff --git a/components/camel-chaos/src/main/docs/chaos.adoc
b/components/camel-chaos/src/main/docs/chaos.adoc
index 54f0b101cc80..3ead495e3743 100644
--- a/components/camel-chaos/src/main/docs/chaos.adoc
+++ b/components/camel-chaos/src/main/docs/chaos.adoc
@@ -171,8 +171,8 @@ only the list and the report.
== Rules
-A rule is one line: the target, the fault and its parameters as `name=value`,
then `at`, `in`, `rate`, `from` and
-`until`, and optionally an expectation. A value with spaces or a `;` is quoted
with `"` or `'` (`\"`, `\'` and `\\` inside quotes
+A rule is one line: the target, the fault and its parameters as `name=value`,
then `at`, `in`, `when`, `rate`,
+`from` and `until`, and optionally an expectation. A value with spaces or a
`;` is quoted with `"` or `'` (`\"`, `\'` and `\\` inside quotes
are a quote and a backslash).
[source]
@@ -182,6 +182,8 @@ fault: error | delay | drop | corrupt | duplicate | stall
| lifecycle, with it
at: send | reply | dlq | poll | entry | step (optional)
in: in=<route id> narrows the sends, polls and steps of endpoint=,
component= and step= to a route
(* matches any characters; optional)
+when: a condition on the message in the simple language, in quotes
(when="${header.country} == 'ES'");
+ the rule fires only for the messages that meet it (optional)
rate: a probability (rate=0.3), every Nth match (rate=every:5), or bursts
(rate=burst:5s/30s: every match
during the first 5s of each 30s); without it every match fires
from: when the rule starts after it is added (from=+30s), such as after the
warm-up; without it at once
@@ -227,6 +229,32 @@ rule is about its route already, and `stall` and
`lifecycle` act on the route of
When no route matches `in=`, or the target matches no send, poll or step
inside those routes, adding the rule warns,
with what is there.
+=== A condition on the message
+
+`when=` fires a rule only for the messages that meet a condition, written in
the
+xref:languages:simple-language.adoc[Simple] language, in quotes:
+
+[source]
+----
+endpoint=http:stock* error status=503 when="${header.country} == 'ES'"
+route=orders error rate=0.1 when="${header.priority} == 'high'"
+endpoint=kafka:orders* duplicate when="${body} contains 'refund'"
+----
+
+The condition is made into a predicate once, when the rule is added (a
condition that is not valid simple is refused,
+with the reason of the simple language), and checked before the rate: `rate`
and an `until` count count only the
+messages that meet it, and the seed gives the same faults for the same
matching messages. "Fail 5% of the orders, or
+the ones from Spain" is two rules, one with `rate=0.05` and one with `when=`.
A condition that fails, such as a method
+call on a header that is not there, is not met: the exchange goes on, and the
report counts it in `skippedReasons`.
+
+A poll has no message, so a rule with `when=` leaves out the polls of its
target, and `at=poll`, `stall` and
+`lifecycle` take no `when=`. A condition that reads the body (`+${body}+`,
`+${bodyAs(...)}+`, `jsonpath`, `jq`, `xpath`)
+reads a stream body from the stream cache, and resets the cache after the
check, so the route reads the whole body; a
+reply that is a stream (`at=reply`) is cached first. Without stream caching a
stream would be consumed, so such a
+condition is refused when stream caching is off. Only simple is supported for
now (a language in front, such as
+`groovy:`, is refused). On the command line, quote the rule with `'` and a
text in the condition with `\"`:
+`+--rule='endpoint=http:stock* error when="${header.country} == \"ES\""'+`.
+
A rule gets an id when it is added (`r1`, `r2`, ...), which change and remove
take. A change replaces the rule with a
new one with the same id and the next version (a rule is version 1 when it is
added), and the replaced one stays in the
report: so the report has an entry per version of the id, and a tool tells
them apart by the `version` of the entry
@@ -646,6 +674,9 @@ seconds
|`route=orders error rate=0.05`
|fail 5% of the messages entering route orders
+|`+endpoint=http:stock* error status=503 when="${header.country} == 'ES'"+`
+|+fail every send to http:stock* with HTTP status 503, when ${header.country}
== 'ES'+
+
|`route=poller lifecycle action=remove duration=30s repeat=3 every=1m`
|remove route poller for 30 seconds, then add it back, 3 times, every 1 minute
|===
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java
index b96a11f80a0a..867de4d2ffde 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java
@@ -137,7 +137,7 @@ public final class ChaosManager {
* expectation is still pending)
*/
public String addRule(String rule) {
- ChaosRule.Builder definition = ChaosRuleParser.parse(rule,
camelContext.getClassResolver(), maxDuration);
+ ChaosRule.Builder definition = parse(rule);
synchronized (lock) {
return add(List.of(definition)).get(0);
}
@@ -160,13 +160,38 @@ public final class ChaosManager {
}
List<ChaosRule.Builder> definitions = new ArrayList<>();
for (String text : texts) {
- definitions.add(ChaosRuleParser.parse(text,
camelContext.getClassResolver(), maxDuration));
+ definitions.add(parse(text));
}
synchronized (lock) {
return add(definitions);
}
}
+ /**
+ * Parses and checks a rule, and makes the predicate of its condition
({@code when=}) with the simple language of
+ * the context, once.
+ */
+ private ChaosRule.Builder parse(String text) {
+ ChaosRule.Builder b = ChaosRuleParser.parse(text,
camelContext.getClassResolver(), maxDuration);
+ if (b.when != null) {
+ if (b.whenReadsBody && !camelContext.isStreamCaching()) {
+ throw new IllegalArgumentException(
+ "Invalid chaos rule [" + b.text + "]: when=" + b.when
+ " reads the body, which consumes a"
+ + " stream body when stream
caching is off (it is on by default)");
+ }
+ try {
+ b.condition =
camelContext.resolveLanguage("simple").createPredicate(b.when);
+ b.condition.init(camelContext);
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException(
+ "Invalid chaos rule [" + b.text + "]: when=" + b.when
+ " is not a valid simple predicate: "
+ + e.getMessage(),
+ e);
+ }
+ }
+ return b;
+ }
+
/** Adds parsed rules, all or none. Called with the lock held. */
private List<String> add(List<ChaosRule.Builder> definitions) {
for (ChaosRule.Builder definition : definitions) {
@@ -215,7 +240,7 @@ public final class ChaosManager {
* @throws IllegalArgumentException when there is no rule with that id, or
the new rule is not valid
*/
public int changeRule(String id, String rule) {
- ChaosRule.Builder definition = ChaosRuleParser.parse(rule,
camelContext.getClassResolver(), maxDuration);
+ ChaosRule.Builder definition = parse(rule);
synchronized (lock) {
int index = indexOf(id);
if (index < 0) {
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java
index 51260d1fad67..80159d7452a2 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java
@@ -311,6 +311,17 @@ final class ChaosModel {
p.put("pick", "routes");
yield p;
}
+ case "when" -> {
+ JsonObject p = param("when", "predicate", false,
+ "A condition on the message in the simple
language, in quotes: the rule fires only for the"
+ + "
messages that meet it, and its rate and until"
+ + " count
only those. Not with at=poll, stall or"
+ + "
lifecycle; a condition on the body needs stream"
+ + "
caching",
+ "\"${header.country} == 'ES'\"");
+ p.put("language", "simple");
+ yield p;
+ }
case "rate" -> {
JsonObject p = param("rate", "rate", false,
"How often the rule fires: a probability above 0
and at most 1, every Nth match, or bursts"
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java
index c0dd6625bffb..a042dc0240d7 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java
@@ -66,7 +66,8 @@ final class ChaosRoutePolicyFactory implements
RoutePolicyFactory {
}
long now = TimeSource.currentTimeMillis();
for (ChaosRule rule : rules) {
- if (rule.live(now) && rule.matchesId(route.getRouteId()) &&
rule.fires(now)) {
+ if (rule.live(now) && rule.matchesId(route.getRouteId()) &&
rule.matchesWhen(exchange)
+ && rule.fires(now)) {
if (rule.fault == ChaosRule.Fault.CORRUPT) {
ChaosFaults.corrupt(rule, exchange, Hook.ENTRY,
route.getRouteId(), null);
} else {
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java
index 8c75615d9414..ce273c6469ab 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.chaos;
+import java.io.InputStream;
+import java.io.Reader;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.List;
@@ -28,7 +30,12 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import org.apache.camel.Exchange;
+import org.apache.camel.Predicate;
+import org.apache.camel.StreamCache;
+import org.apache.camel.support.MessageHelper;
import org.apache.camel.util.TimeSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A rule that was added: what it targets and does (parsed by {@link
ChaosRuleParser}, immutable), and its state while
@@ -136,6 +143,11 @@ final class ChaosRule {
}
}
+ private static final Logger LOG = LoggerFactory.getLogger(ChaosRule.class);
+
+ // the reason of a skip when the condition (when=) threw, one of the fixed
texts of the skipped reasons
+ static final String CONDITION_FAILED = "the condition (when=) failed with
an exception";
+
// the golden ratio increment of SplitMix64
private static final long GOLDEN_GAMMA = 0x9e3779b97f4a7c15L;
@@ -151,6 +163,11 @@ final class ChaosRule {
final String pattern;
// the routes the rule is narrowed to (in=, a route id or a pattern), else
null
final String in;
+ // the condition on the message (when=, in the simple language), else
null; the predicate is made when the rule is
+ // added, so it is null in a rule made without a context
+ final String when;
+ private final Predicate condition;
+ private final boolean conditionReadsBody;
// the endpoint= pattern without the // after its scheme, made once (the
hooks match it on every send and poll)
private final String uriPattern;
final Set<Hook> hooks;
@@ -234,6 +251,9 @@ final class ChaosRule {
this.target = b.target;
this.pattern = b.pattern;
this.in = b.in;
+ this.when = b.when;
+ this.condition = b.condition;
+ this.conditionReadsBody = b.whenReadsBody;
this.uriPattern = target == Target.ENDPOINT ?
ChaosPatterns.withoutSlashes(pattern) : null;
this.hooks = Set.copyOf(b.hooks);
this.fault = b.fault;
@@ -469,6 +489,9 @@ final class ChaosRule {
} else {
sb.append(verb()).append(' ').append(share(at));
sb.append(faultWords());
+ if (when != null) {
+ sb.append(", when ").append(when);
+ }
if (burstPeriod > 0) {
sb.append(", during the first
").append(ChaosWords.time(burstOn)).append(" of every ")
.append(ChaosWords.time(burstPeriod));
@@ -746,6 +769,44 @@ final class ChaosRule {
return in == null || matchesIn(ChaosExchangeState.routeId(exchange));
}
+ /**
+ * Whether the message meets the condition of {@code when=} (true for a
rule without one). A condition that throws
+ * is not met, and counts as skipped. A condition on the body reads a
stream body from its cache: a stream that is
+ * not cached yet (such as the reply of a send) is cached first, and the
cache is reset after the check, so the
+ * route reads the body as before.
+ */
+ boolean matchesWhen(Exchange exchange) {
+ Predicate c = condition;
+ if (c == null) {
+ return true;
+ }
+ try {
+ if (conditionReadsBody) {
+ cacheStreamBody(exchange);
+ }
+ return c.matches(exchange);
+ } catch (Exception e) {
+ LOG.debug("The condition of chaos rule {} (when={}) failed on
exchange {}: {}", id, when,
+ exchange.getExchangeId(), e.getMessage(), e);
+ skipped(CONDITION_FAILED);
+ return false;
+ } finally {
+ if (conditionReadsBody) {
+ MessageHelper.resetStreamCache(exchange.getMessage());
+ }
+ }
+ }
+
+ private static void cacheStreamBody(Exchange exchange) {
+ Object body = exchange.getMessage().getBody();
+ if ((body instanceof InputStream || body instanceof Reader) && !(body
instanceof StreamCache)) {
+ StreamCache cache =
exchange.getContext().getStreamCachingStrategy().cache(exchange);
+ if (cache != null) {
+ exchange.getMessage().setBody(cache);
+ }
+ }
+ }
+
/** Whether the route is one of {@code in=} (true for a rule without it).
*/
boolean matchesIn(String routeId) {
return in == null || (routeId != null && ChaosPatterns.matches(in,
routeId));
@@ -1004,6 +1065,9 @@ final class ChaosRule {
Target target;
String pattern;
String in;
+ String when;
+ boolean whenReadsBody;
+ Predicate condition;
Hook at;
Set<Hook> hooks;
Fault fault;
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java
index eb695dd9e157..e7ed74c0d784 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java
@@ -57,7 +57,7 @@ final class ChaosRuleParser {
// the options of a rule besides the parameters of its fault: all of them
for the faults at a hook, and those of a
// stall or lifecycle rule, which acts on its route at no hook
- static final Set<String> GENERAL_KEYS = Set.of("at", "in", "rate",
"until", "from", "expect", "within");
+ static final Set<String> GENERAL_KEYS = Set.of("at", "in", "when", "rate",
"until", "from", "expect", "within");
static final Set<String> LIFECYCLE_KEYS = Set.of("from", "expect",
"within");
// the prefix of the response headers of error status=:
header.<name>=<value>
static final String HEADER_PREFIX = "header.";
@@ -69,9 +69,6 @@ final class ChaosRuleParser {
private static final Set<String> FAULT_KEYS
= Set.of("exception", "message", "status", "body", "after",
"duration", "mode", "header", "action", "repeat",
"every");
- // parameters of the design that are not supported yet, with what to use
instead
- private static final Map<String, String> NOT_SUPPORTED = Map.of(
- "when", "when= (a condition on the message) is not supported yet;
a rule fires by rate= and ends by until=");
private ChaosRuleParser() {
}
@@ -211,7 +208,7 @@ final class ChaosRuleParser {
String unknown = null;
for (String key : params.keySet()) {
if (unknown == null && !GENERAL_KEYS.contains(key) &&
!FAULT_KEYS.contains(key)
- && !key.startsWith(HEADER_PREFIX) &&
!NOT_SUPPORTED.containsKey(key)) {
+ && !key.startsWith(HEADER_PREFIX)) {
unknown = key;
}
}
@@ -233,6 +230,11 @@ final class ChaosRuleParser {
}
from(b, params.get("from"), maxDuration);
in(b, params.get("in"));
+ if (b.fault == Fault.LIFECYCLE && params.containsKey("when")) {
+ throw new IllegalArgumentException(
+ "when= is a condition on the message, but the fault " +
(b.stall ? "stall" : "lifecycle")
+ + " acts on the route, with no
message");
+ }
if (b.fault == Fault.LIFECYCLE) {
// stall and lifecycle act on the route: no hook, rate or until
lifecycle(b, params, maxDuration);
@@ -299,6 +301,7 @@ final class ChaosRuleParser {
}
hooks(b, params.get("at"), http);
+ when(b, params.get("when"));
if (b.after > 0 && b.at == Hook.REPLY) {
// the reply hook runs on the thread that completes the send,
which can be an I/O thread of the component
throw new IllegalArgumentException(
@@ -475,10 +478,6 @@ final class ChaosRuleParser {
"action= is a parameter of the fault lifecycle
(route=<id> lifecycle action=suspend|stop|remove"
+ " duration=<time>), not
of the fault " + fault);
}
- String notSupported = NOT_SUPPORTED.get(key);
- if (notSupported != null) {
- throw new IllegalArgumentException(notSupported);
- }
StringJoiner known = new StringJoiner(", ");
faultKeys.stream().sorted().forEach(k ->
known.add(k.equals(HEADER_PREFIX) ? "header.<name>" : k));
generalKeys.stream().sorted().forEach(known::add);
@@ -986,6 +985,73 @@ final class ChaosRuleParser {
}
}
+ /**
+ * A condition on the message ({@code when="${header.country} == 'ES'"}),
in the simple language: the rule fires
+ * only for the messages that meet it. It is checked as text here, and
made into a predicate when the rule is added
+ * (the language of the context). A poll has no message, so a rule with a
condition leaves out the polls of its
+ * target, and is refused with {@code at=poll}.
+ */
+ private static void when(ChaosRule.Builder b, String when) {
+ if (when == null) {
+ return;
+ }
+ if (languagePrefix(when) != null) {
+ throw new IllegalArgumentException(
+ "when= takes a condition in the simple language, without a
language in front ("
+ + languagePrefix(when) + ": is
not supported), such as"
+ + " when=\"${header.country} ==
'ES'\"");
+ }
+ if (b.at == Hook.POLL) {
+ throw new IllegalArgumentException(
+ "when= is a condition on the message, but a poll (at=poll)
has no message");
+ }
+ b.hooks.remove(Hook.POLL);
+ if (b.hooks.isEmpty()) {
+ throw new IllegalArgumentException(
+ "when= is a condition on the message, and the rule fires
only at polls, which have no message");
+ }
+ b.when = when;
+ b.whenReadsBody = readsBody(when);
+ }
+
+ // a language in front of the condition, such as groovy: (letters, digits
and - before a colon at the start), or null
+ private static String languagePrefix(String when) {
+ int i = 0;
+ while (i < when.length()) {
+ char ch = when.charAt(i);
+ if (!(ch >= 'a' && ch <= 'z' || ch >= 'A' && ch <= 'Z' || i > 0 &&
(ch >= '0' && ch <= '9' || ch == '-'))) {
+ break;
+ }
+ i++;
+ }
+ return i > 0 && i < when.length() && when.charAt(i) == ':' ?
when.substring(0, i) : null;
+ }
+
+ /**
+ * Whether a condition reads the body of the message: a word of it names
the body (${body}, ${bodyAs(String)},
+ * ${in.body}, ${mandatoryBodyAs(String)}, ${prettyBody}, not ${bodyType}
or ${header.somebody}), or it uses a
+ * language of the body (jsonpath, jq, xpath). Reading a stream body
consumes it, so when in doubt it says yes: that
+ * only turns on stream caching.
+ */
+ static boolean readsBody(String when) {
+ int i = 0;
+ while (i < when.length()) {
+ if (!Character.isLetterOrDigit(when.charAt(i))) {
+ i++;
+ continue;
+ }
+ int start = i;
+ while (i < when.length() &&
Character.isLetterOrDigit(when.charAt(i))) {
+ i++;
+ }
+ String word = when.substring(start, i);
+ if ((word.startsWith("body") || word.contains("Body")) &&
!word.equals("bodyType")) {
+ return true;
+ }
+ }
+ return when.contains("jsonpath(") || when.contains("jq(") ||
when.contains("xpath(");
+ }
+
/**
* The routes a rule is narrowed to ({@code in=orders}, or a pattern such
as {@code in=order*}): the sends, replies,
* steps and polls of its target in those routes only. A route is the
target of {@code route=} already, and
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java
index 5a84ca2ee0ee..628be5f64a2e 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java
@@ -69,7 +69,8 @@ final class ChaosSendHook implements SendHook, Ordered {
long now = TimeSource.currentTimeMillis();
String uri = ChaosPatterns.endpointUri(endpoint);
for (ChaosRule rule : rules) {
- if (rule.live(now) && rule.matchesUri(uri) &&
rule.matchesIn(exchange) && rule.fires(now)) {
+ if (rule.live(now) && rule.matchesUri(uri) &&
rule.matchesIn(exchange) && rule.matchesWhen(exchange)
+ && rule.fires(now)) {
if (deadLetter && (rule.fault == ChaosRule.Fault.ERROR ||
rule.fault == ChaosRule.Fault.DROP)) {
// the message does not reach the dead letter channel:
lost, when the channel then handles it
state.deadLetterFault = rule;
@@ -118,7 +119,8 @@ final class ChaosSendHook implements SendHook, Ordered {
long now = TimeSource.currentTimeMillis();
String uri = ChaosPatterns.endpointUri(endpoint);
for (ChaosRule rule : rules) {
- if (rule.live(now) && rule.matchesUri(uri) &&
rule.matchesIn(exchange) && rule.fires(now)) {
+ if (rule.live(now) && rule.matchesUri(uri) &&
rule.matchesIn(exchange) && rule.matchesWhen(exchange)
+ && rule.fires(now)) {
ChaosFaults.atReply(rule, exchange, endpoint);
return;
}
diff --git
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java
index 641eab60963b..747c5ae3e49d 100644
---
a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java
+++
b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java
@@ -53,7 +53,8 @@ final class ChaosStepHook implements StepHook, Ordered {
}
long now = TimeSource.currentTimeMillis();
for (ChaosRule rule : rules) {
- if (rule.live(now) && rule.matchesId(nodeId) &&
rule.matchesIn(exchange) && rule.fires(now)) {
+ if (rule.live(now) && rule.matchesId(nodeId) &&
rule.matchesIn(exchange) && rule.matchesWhen(exchange)
+ && rule.fires(now)) {
rule.fired(Hook.STEP, exchange, null, nodeId);
return fault(rule, exchange);
}
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosMainTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosMainTest.java
index e86c917ca0be..76466184c1fb 100644
---
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosMainTest.java
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosMainTest.java
@@ -79,6 +79,17 @@ class ChaosMainTest {
assertThat(send().getException()).isInstanceOf(ChaosException.class).hasMessage("boom");
}
+ @Test
+ void aConditionInTheRulesOfTheProperties() throws Exception {
+ main().addInitialProperty("camel.main.profile", "dev");
+ main.addInitialProperty("camel.chaos.rules", "endpoint=mock:target
error when=\"${header.country} == 'ES'\"");
+ main.start();
+
+ assertThat(send().getException()).isNull();
+ Exchange es = main.getCamelTemplate().send("direct:start", e ->
e.getMessage().setHeader("country", "ES"));
+ assertThat(es.getException()).isInstanceOf(ChaosException.class);
+ }
+
@Test
void profileProdRefusesToStartWithRules() {
main().addInitialProperty("camel.main.profile", "prod");
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java
index 205c803c6d4e..142d027fc02d 100644
---
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java
@@ -100,7 +100,8 @@ class ChaosModelTest {
assertThat(strings(byName("faults",
"duplicate").get("hooks"))).containsExactly("send");
assertThat(strings(byName("faults",
"duplicate").get("targets"))).containsExactly("component", "endpoint");
assertThat(paramNames(byName("faults", "duplicate"))).isEmpty();
- assertThat(names("options")).containsExactly("at", "expect", "from",
"in", "rate", "until", "within");
+ assertThat(names("options")).containsExactly("at", "expect", "from",
"in", "rate", "until", "when",
+ "within");
// the TUI form picks the route of in=
assertThat(byName("options",
"in").getString("pick")).isEqualTo("routes");
}
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java
index 6848fb942193..4d62f48c992d 100644
---
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java
@@ -78,6 +78,8 @@ class ChaosRuleDescribeTest {
route=orders error rate=0.05 | fail 5% of the messages
entering route orders
route=orders error exception=java.net.ConnectException
| fail every message entering route orders with ConnectException
route=order* delay duration=2s until=10 | delay every
message entering routes order* by 2 seconds, for the next 10 messages
+ endpoint=http:stock* error status=503
when="${header.country} == 'ES'" | fail every send to http:stock* with HTTP
status 503, when ${header.country} == 'ES'
+ route=orders error rate=0.05 when='${header.vip}'
until=10m | fail 5% of the messages entering route orders, when ${header.vip},
for 10 minutes
route=poller stall duration=20s | suspend route poller
for 20 seconds
route=poller lifecycle action=stop duration=1m30s |
stop route poller for 1 minute 30 seconds
route=poller lifecycle action=remove duration=30s
repeat=3 every=1m from=+10s | remove route poller for 30 seconds, then add it
back, 3 times, every 1 minute, starting after 10 seconds
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java
index cf6503e4ad92..c776ad4e0201 100644
---
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java
@@ -143,6 +143,36 @@ class ChaosRuleParserTest {
assertThat(parse("endpoint=x error").in).isNull();
}
+ @Test
+ void whenIsAConditionOnTheMessage() {
+ ChaosRule.Builder b = parse("endpoint=http:stock* error status=503
when=\"${header.country} == 'ES'\"");
+ assertThat(b.when).isEqualTo("${header.country} == 'ES'");
+ assertThat(b.whenReadsBody).isFalse();
+ // a poll has no message: without at=, the rule leaves out the polls
of its target
+ assertThat(parse("endpoint=x error when=\"${header.a} ==
1\"").hooks).containsExactly(Hook.SEND);
+ assertThat(parse("route=orders error when=\"${header.a} ==
1\"").hooks).containsExactly(Hook.ENTRY);
+ assertThat(parse("step=s1 delay duration=1s when='${header.a} ==
1'").when).isEqualTo("${header.a} == 1");
+ // the predicate is made when the rule is added
+ assertThat(b.condition).isNull();
+ // a time such as 08:00 is not a language
+ assertThat(parse("endpoint=x error when=\"${date:now:HH:mm} ==
'08:00'\"").when).startsWith("${date");
+ }
+
+ @Test
+ void aConditionThatReadsTheBody() {
+ assertThat(ChaosRuleParser.readsBody("${body} contains
'boom'")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${bodyAs(String)} ==
'x'")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${in.body.length} >
10")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${jsonpath($.country)} ==
'ES'")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${mandatoryBodyAs(String)} ==
'x'")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${prettyBody} contains
'x'")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${bodyOneLine} contains
'x'")).isTrue();
+ assertThat(ChaosRuleParser.readsBody("${bodyType} ==
'java.lang.String'")).isFalse();
+ assertThat(ChaosRuleParser.readsBody("${header.country} ==
'ES'")).isFalse();
+ assertThat(ChaosRuleParser.readsBody("${header.somebody} ==
'x'")).isFalse();
+ assertThat(ChaosRuleParser.readsBody("${header.antibody} == 'x' &&
${bodyType} != null")).isFalse();
+ }
+
@Test
void errorAndDelayAtTheEntryOfARoute() {
ChaosRule.Builder error = parse("route=orders error
exception=java.net.ConnectException rate=0.05");
@@ -317,7 +347,12 @@ class ChaosRuleParserTest {
"error|needs a target",
"rout=orders error|a rule needs a target: route=<id>,
endpoint=<uri>, component=<name> or step=<id>"
+ " (rout= is not a target)",
- "endpoint=x error when=x|when= (a condition on the message) is not
supported yet",
+ "endpoint=x error at=poll when=\"${header.a} == 1\"|when= is a
condition on the message, but a poll (at=poll) has"
+ + " no
message",
+ "route=orders stall duration=1m when=x|when= is a condition on the
message, but the fault stall acts on the"
+
+ " route, with no message",
+ "endpoint=x error when=\"groovy:headers.a == 1\"|when= takes a
condition in the simple language, without a"
+
+ " language in front (groovy: is not
supported)",
"endpoint=x error action=stop|action= is a parameter of the fault
lifecycle (route=<id> lifecycle action=",
"component=kafka error status=503|(http, https, netty-http,
undertow, vertx-http), not for kafka",
"route=orders drop|drop is a fault of the sends and polls: target
the sends of the route with endpoint=*"
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosWhenNoStreamCachingTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosWhenNoStreamCachingTest.java
new file mode 100644
index 000000000000..2cc9f457200d
--- /dev/null
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosWhenNoStreamCachingTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.component.chaos;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Without stream caching a condition that reads the body would consume a
stream body, so it is refused; one on the
+ * headers is fine.
+ */
+class ChaosWhenNoStreamCachingTest extends ChaosTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.setStreamCaching(false);
+ return camelContext;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:orders").routeId("orders").to("mock:stock");
+ }
+ };
+ }
+
+ @Test
+ void aConditionOnTheBodyIsRefused() {
+ assertThatThrownBy(() -> chaos().addRule("endpoint=mock:stock error
when=\"${body} contains 'boom'\""))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageEndingWith("when=${body} contains 'boom' reads the
body, which consumes a stream body when"
+ + " stream caching is off (it is on by
default)");
+ assertThat(chaos().addRule("endpoint=mock:stock error
when=\"${header.country} == 'ES'\"")).isEqualTo("r1");
+ }
+}
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosWhenTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosWhenTest.java
new file mode 100644
index 000000000000..0647bdaa6afb
--- /dev/null
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosWhenTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.component.chaos;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * {@code when=}: a rule fires only for the messages that meet a condition in
the simple language, checked before the
+ * rate, so the rate and the until count count only those messages.
+ */
+class ChaosWhenTest extends ChaosTestSupport {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:orders").routeId("orders")
+ .setHeader("seen", constant("yes")).id("mark")
+ .to("mock:stock").id("callStock");
+ from("direct:reply").routeId("reply")
+ .to("mock:service").id("callService")
+ .to("mock:after").id("after");
+ }
+ };
+ }
+
+ private Exchange order(String country) {
+ return template.send("direct:orders", e -> {
+ e.getMessage().setHeader("country", country);
+ e.getMessage().setBody("order");
+ });
+ }
+
+ @Test
+ void onlyTheMessagesThatMeetTheCondition() throws Exception {
+ String id = chaos().addRule("endpoint=mock:stock error
when=\"${header.country} == 'ES'\"");
+
getMockEndpoint("mock:stock").expectedHeaderValuesReceivedInAnyOrder("country",
"DK", "SE");
+
+ assertThat(order("DK").getException()).isNull();
+
assertThat(order("ES").getException()).isInstanceOf(ChaosException.class);
+ assertThat(order("SE").getException()).isNull();
+
+ MockEndpoint.assertIsSatisfied(context);
+ // the messages that do not meet the condition are not matches
+ assertThat(ruleReport(id).getLong("matches")).isEqualTo(1);
+ assertThat(fired(id, "send")).isEqualTo(1);
+ assertThat(ruleReport(id).getString("description"))
+ .isEqualTo("fail every send to mock:stock, when
${header.country} == 'ES'");
+ }
+
+ @Test
+ void doubleQuotesInsideEaseTheQuotingOnTheCommandLine() {
+ // camel cmd chaos-add --rule='endpoint=mock:stock error
when="${header.country} == \"ES\""'
+ String id = chaos().addRule("endpoint=mock:stock error
when=\"${header.country} == \\\"ES\\\"\"");
+
+ assertThat(order("DK").getException()).isNull();
+
assertThat(order("ES").getException()).isInstanceOf(ChaosException.class);
+
+ assertThat(fired(id, "send")).isEqualTo(1);
+ // a text needs quotes in the simple language, which says so
+ assertThatThrownBy(() -> chaos().addRule("endpoint=mock:stock error
when=\"${header.country} == ES\""))
+ .hasMessageContaining("write it as a quoted literal 'x'");
+ }
+
+ @Test
+ void theRateCountsOnlyTheMessagesThatMeetIt() {
+ String id = chaos().addRule("endpoint=mock:stock error rate=every:2
when=\"${header.country} == 'ES'\"");
+
+ order("DK");
+ order("DK");
+ assertThat(order("ES").getException()).isNull();
+ order("DK");
+
assertThat(order("ES").getException()).isInstanceOf(ChaosException.class);
+
+ assertThat(ruleReport(id).getLong("matches")).isEqualTo(2);
+ }
+
+ @Test
+ void untilCountsOnlyTheMessagesThatMeetIt() {
+ String id = chaos().addRule("endpoint=mock:stock error until=1
when=\"${header.country} == 'ES'\"");
+
+ order("DK");
+
assertThat(order("ES").getException()).isInstanceOf(ChaosException.class);
+ assertThat(order("ES").getException()).isNull();
+
+ assertThat(ruleReport(id).getString("state")).isEqualTo("ended");
+ }
+
+ @Test
+ void atTheEntryOfARouteAndAtAStep() {
+ String entry = chaos().addRule("route=orders error
when=\"${header.country} == 'ES'\"");
+ String step = chaos().addRule("step=mark delay duration=1ms
when=\"${header.country} == 'DK'\"");
+
+
assertThat(order("ES").getException()).isInstanceOf(ChaosException.class);
+ assertThat(order("DK").getException()).isNull();
+ assertThat(order("SE").getException()).isNull();
+
+ assertThat(fired(entry, "entry")).isEqualTo(1);
+ assertThat(fired(step, "step")).isEqualTo(1);
+ }
+
+ @Test
+ void aConditionOnAStreamBodyLeavesTheBodyToTheRoute() throws Exception {
+ String id = chaos().addRule("endpoint=mock:stock error when=\"${body}
contains 'boom'\"");
+ getMockEndpoint("mock:stock").expectedBodiesReceived("all is well");
+
+ Exchange boom = template.send("direct:orders",
+ e -> e.getMessage().setBody(new
ByteArrayInputStream("boom".getBytes(StandardCharsets.UTF_8))));
+ template.send("direct:orders",
+ e -> e.getMessage().setBody(new ByteArrayInputStream("all is
well".getBytes(StandardCharsets.UTF_8))));
+
+ assertThat(boom.getException()).isInstanceOf(ChaosException.class);
+ // the stream cache was reset after the check, so the endpoint reads
the whole body
+ MockEndpoint.assertIsSatisfied(context);
+ assertThat(fired(id, "send")).isEqualTo(1);
+ }
+
+ @Test
+ void aConditionOnAStreamReplyCachesItFirst() throws Exception {
+ getMockEndpoint("mock:service").whenAnyExchangeReceived(e ->
e.getMessage().setBody(
+ new
ByteArrayInputStream(e.getMessage().getBody(String.class).getBytes(StandardCharsets.UTF_8))));
+ String id = chaos().addRule("endpoint=mock:service error at=reply
when=\"${body} contains 'bad'\"");
+ getMockEndpoint("mock:after").expectedBodiesReceived("good reply");
+
+ assertThat(template.request("direct:reply", e ->
e.getMessage().setBody("bad reply")).getException())
+ .isInstanceOf(ChaosException.class);
+ assertThat(template.request("direct:reply", e ->
e.getMessage().setBody("good reply")).getException()).isNull();
+
+ // the reply stream was cached for the check, and the next step reads
it whole
+ MockEndpoint.assertIsSatisfied(context);
+ assertThat(fired(id, "reply")).isEqualTo(1);
+ }
+
+ @Test
+ void aConditionThatFailsIsNotMetAndCountsAsSkipped() {
+ String id = chaos().addRule("endpoint=mock:stock error
when=\"${header.country.substring(10)} == 'x'\"");
+
+ assertThat(order("ES").getException()).isNull();
+
+ assertThat(ruleReport(id).getLong("matches")).isZero();
+ assertThat(ruleReport(id).getLong("skipped")).isEqualTo(1);
+ JsonObject reasons = (JsonObject) ruleReport(id).get("skippedReasons");
+ assertThat(reasons.getLong(ChaosRule.CONDITION_FAILED)).isEqualTo(1);
+ }
+
+ @Test
+ void aConditionThatIsNotSimpleIsRefused() {
+ assertThatThrownBy(() -> chaos().addRule("endpoint=mock:stock error
when=\"${header.country == 'ES'\""))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageStartingWith("Invalid chaos rule
[endpoint=mock:stock error when=\"${header.country == 'ES'\"]:"
+ + " when=${header.country == 'ES' is
not a valid simple predicate: ");
+ assertThat((JsonArray)
chaos().report().toJson().get("rules")).isEmpty();
+ }
+}
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java
index 7ecc712d4b36..6873634335e3 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java
@@ -44,6 +44,7 @@ import picocli.CommandLine;
" camel cmd chaos-add myApp --rule='step=enrich*
error exception=java.sql.SQLException rate=0.1'",
" camel cmd chaos-add myApp
--rule='endpoint=http:* in=orders error status=503 rate=0.1'",
" camel cmd chaos-add myApp --rule='route=orders
error rate=0.05'",
+ " camel cmd chaos-add myApp
--rule='endpoint=http:stock* error when=\"$${header.country} == \\\"ES\\\"\"'",
" camel cmd chaos-add myApp
--rule='endpoint=seda:dlq error at=dlq' --rule='endpoint=http:stock* error'" })
public class CamelChaosAddAction extends CamelChaosAction {
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 6af350b61372..2db0b0231723 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
@@ -675,6 +675,8 @@ SQLException rate=0.1'
camel cmd chaos-add myApp --rule='endpoint=http:* in=orders error status=503
rate=0.1'
camel cmd chaos-add myApp --rule='route=orders error rate=0.05'
+ camel cmd chaos-add myApp --rule='endpoint=http:stock* error when="${header.
+country} == \"ES\""'
camel cmd chaos-add myApp --rule='endpoint=seda:dlq error at=dlq'
--rule='endpoint=http:stock* error'
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java
index 6fb2106f54f1..997dc728130b 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java
@@ -118,7 +118,7 @@ final class ChaosInfo {
/**
* The fault of the rule with its parameters, as it is written,
without the target and the options of the rule
- * (at, in, rate, until, from, expect, within), such as {@code error
status=503}.
+ * (at, in, when, rate, until, from, expect, within), such as {@code
error status=503}.
*/
String fault() {
StringBuilder sb = new StringBuilder();
@@ -242,7 +242,7 @@ final class ChaosInfo {
}
// the options of a rule besides its target and its fault
- private static final List<String> OPTIONS = List.of("at", "in", "rate",
"until", "from", "expect", "within");
+ private static final List<String> OPTIONS = List.of("at", "in", "when",
"rate", "until", "from", "expect", "within");
/** The counts of a route in the report. */
record RouteCounts(String route, long completed, long failed, long
failedKept, long handled, long redeliveries) {
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java
index a8bf18aa3734..649c9747a514 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java
@@ -307,5 +307,9 @@ class ChaosTabRenderTest {
steps.text = "step=* in=order* delay duration=1s";
assertThat(ChaosTab.target(steps)).isEqualTo("step * in order*");
assertThat(steps.fault()).isEqualTo("delay duration=1s");
+ // a condition (when=) is an option, not a parameter of the fault
+ ChaosInfo.Rule when = new ChaosInfo.Rule();
+ when.text = "endpoint=http:stock* error status=503
when=\"${header.country} == 'ES'\" rate=0.5";
+ assertThat(when.fault()).isEqualTo("error status=503");
}
}
