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 72a1bff606a3 CAMEL-25928: runtime checks of the body types of the EIPs
implemented outside camel-core (#28451)
72a1bff606a3 is described below
commit 72a1bff606a3349674d027368132e30fe02aebc0
Author: Claus Ibsen <[email protected]>
AuthorDate: Sun Oct 25 15:28:01 2026 +0100
CAMEL-25928: runtime checks of the body types of the EIPs implemented
outside camel-core (#28451)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../component/a2a/A2ASubTaskBodyTypesTest.java | 94 ++++++++++++
.../tokenizer/TokenizerBodyTypesTest.java | 99 ++++++++++++
.../component/kamelet/KameletBodyTypesTest.java | 90 +++++++++++
.../FaultToleranceBodyTypesTest.java | 165 ++++++++++++++++++++
.../resilience4j/CircuitBreakerBodyTypesTest.java | 170 +++++++++++++++++++++
5 files changed, 618 insertions(+)
diff --git
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/A2ASubTaskBodyTypesTest.java
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/A2ASubTaskBodyTypesTest.java
new file mode 100644
index 000000000000..de51d207b681
--- /dev/null
+++
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/A2ASubTaskBodyTypesTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.a2a;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.A2ASubTaskDefinition;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The body types of the a2aSubTask EIP (CAMEL-25928), checked at runtime: its
steps start with the body as it was, and
+ * the body after it is what its steps end with.
+ */
+class A2ASubTaskBodyTypesTest extends CamelTestSupport {
+
+ /** The body the steps end with. */
+ private static final String END = "end of the steps";
+
+ private final List<Object> inside = new CopyOnWriteArrayList<>();
+
+ A2ASubTaskBodyTypesTest() {
+ testConfiguration().withUseRouteBuilder(false);
+ }
+
+ @Test
+ void theStepsStartWithTheBodyAndTheBodyAfterIsWhatTheyEndWith() throws
Exception {
+ BodyTypes bodyTypes =
A2ASubTaskDefinition.class.getAnnotation(BodyTypes.class);
+ assertThat(bodyTypes).isNotNull();
+ assertThat(bodyTypes.eipOut().from()).isEqualTo("steps");
+
assertThat(bodyTypes.eipSteps().value()).containsExactly(Carrier.UNCHANGED);
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:end")
+ .a2aSubTask()
+ .emitBefore("Before ${body}")
+ .emitAfter("After ${body}")
+ .process(e -> inside.add(e.getMessage().getBody()))
+ .setBody(constant(END))
+ .end();
+
+ from("direct:keep")
+ .a2aSubTask()
+ .emitBefore("Before ${body}")
+ .emitAfter("After ${body}")
+ .process(e -> inside.add(e.getMessage().getBody()))
+ .end();
+ }
+ });
+ context.start();
+
+ Order body = new Order();
+ Exchange out = template.send("direct:end", e ->
e.getMessage().setBody(body));
+ assertThat(out.getException()).isNull();
+ assertThat(inside).hasSize(1);
+ assertThat(inside.get(0)).isSameAs(body);
+ assertThat(out.getMessage().getBody()).isEqualTo(END);
+
+ // steps that leave the body as it was give the same instance after
the EIP
+ inside.clear();
+ out = template.send("direct:keep", e -> e.getMessage().setBody(body));
+ assertThat(out.getException()).isNull();
+ assertThat(inside).hasSize(1);
+ assertThat(inside.get(0)).isSameAs(body);
+ assertThat(out.getMessage().getBody()).isSameAs(body);
+ }
+
+ /** A body that no converter knows: an EIP that keeps it gives the same
instance. */
+ private static final class Order {
+ }
+}
diff --git
a/components/camel-ai/camel-langchain4j-tokenizer/src/test/java/org/apache/camel/component/langchain4j/tokenizer/TokenizerBodyTypesTest.java
b/components/camel-ai/camel-langchain4j-tokenizer/src/test/java/org/apache/camel/component/langchain4j/tokenizer/TokenizerBodyTypesTest.java
new file mode 100644
index 000000000000..239e87699aea
--- /dev/null
+++
b/components/camel-ai/camel-langchain4j-tokenizer/src/test/java/org/apache/camel/component/langchain4j/tokenizer/TokenizerBodyTypesTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.langchain4j.tokenizer;
+
+import java.nio.charset.StandardCharsets;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.TokenizerDefinition;
+import org.apache.camel.model.tokenizer.LangChain4jTokenizerDefinition;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The body types of the tokenizer EIP (CAMEL-25928), checked at runtime: it
reads the body as text and leaves the
+ * chunks of the text as a String[].
+ */
+public class TokenizerBodyTypesTest extends LangChain4JTokenizerTestSupport {
+
+ @Test
+ public void textGivesTheChunksAsAnArray() {
+ BodyTypes bodyTypes =
TokenizerDefinition.class.getAnnotation(BodyTypes.class);
+ assertNotNull(bodyTypes);
+ assertEquals(Carrier.TEXT, bodyTypes.eipIn().value()[0]);
+ BodyType out = bodyTypes.eipOut();
+ assertEquals(Carrier.LIST, out.value()[0]);
+ assertEquals("java.lang.String[]", out.javaType());
+
+ Exchange exchange = template.send("direct:start", e ->
e.getMessage().setBody(TEXT));
+ assertNull(exchange.getException());
+ Object body = exchange.getMessage().getBody();
+ assertInstanceOf(String[].class, body);
+ assertEquals(out.javaType(), body.getClass().getCanonicalName());
+ assertEquals(4, ((String[]) body).length);
+ }
+
+ @Test
+ public void bytesAreReadAsText() {
+ // the tokenizer reads the body with getBody(String.class), so bytes
are converted to text first
+ Exchange exchange
+ = template.send("direct:start", e ->
e.getMessage().setBody(TEXT.getBytes(StandardCharsets.UTF_8)));
+ assertNull(exchange.getException());
+ Exchange text = template.send("direct:start", e ->
e.getMessage().setBody(TEXT));
+ assertArrayEquals((String[]) text.getMessage().getBody(), (String[])
exchange.getMessage().getBody());
+ }
+
+ @Test
+ public void anObjectNoConverterKnowsIsReadAsItsToString() {
+ // no converter knows Order, but the fallback converter to String uses
toString(), so the object is not
+ // refused: its toString() is what is split into chunks
+ Order order = new Order();
+ Exchange exchange = template.send("direct:start", e ->
e.getMessage().setBody(order));
+ assertNull(exchange.getException());
+ assertArrayEquals(new String[] { order.toString() }, (String[])
exchange.getMessage().getBody());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .tokenize(tokenizer()
+ .byParagraph()
+ .maxSegmentSize(1024)
+ .maxOverlap(10)
+
.using(LangChain4jTokenizerDefinition.TokenizerType.OPEN_AI)
+ .end());
+ }
+ };
+ }
+
+ /** A body that no converter knows. */
+ private static final class Order {
+ }
+}
diff --git
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletBodyTypesTest.java
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletBodyTypesTest.java
new file mode 100644
index 000000000000..65d40d056e98
--- /dev/null
+++
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletBodyTypesTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.kamelet;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.KameletDefinition;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The body types of the kamelet EIP (CAMEL-25928), checked at runtime. The
catalog says UNKNOWN for the body after it,
+ * as it is what the route of the Kamelet leaves: the same EIP gives a String
with one Kamelet, keeps the body with
+ * another and gives a List with a third.
+ */
+public class KameletBodyTypesTest extends CamelTestSupport {
+
+ @Test
+ public void theBodyAfterIsWhatTheKameletRouteLeaves() {
+ BodyTypes bodyTypes =
KameletDefinition.class.getAnnotation(BodyTypes.class);
+ assertThat(bodyTypes).isNotNull();
+ BodyType out = bodyTypes.eipOut();
+ assertThat(out.value()).containsExactly(Carrier.UNKNOWN);
+ assertThat(out.from()).isEmpty();
+
+ Order body = new Order();
+
+ Exchange text = template.send("direct:text", e ->
e.getMessage().setBody(body));
+ assertThat(text.getException()).isNull();
+ assertThat(text.getMessage().getBody()).isEqualTo("Hello from the
Kamelet");
+
+ Exchange keep = template.send("direct:keep", e ->
e.getMessage().setBody(body));
+ assertThat(keep.getException()).isNull();
+ assertThat(keep.getMessage().getBody()).isSameAs(body);
+
+ Exchange list = template.send("direct:list", e ->
e.getMessage().setBody(body));
+ assertThat(list.getException()).isNull();
+ assertThat(list.getMessage().getBody()).isEqualTo(List.of("a", "b"));
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("text")
+ .from("kamelet:source")
+ .setBody(constant("Hello from the Kamelet"));
+
+ routeTemplate("keep")
+ .from("kamelet:source")
+ .setHeader("visited", constant(true));
+
+ routeTemplate("list")
+ .from("kamelet:source")
+ .setBody(constant(List.of("a", "b")));
+
+ from("direct:text").kamelet("text");
+ from("direct:keep").kamelet("keep");
+ from("direct:list").kamelet("list");
+ }
+ };
+ }
+
+ /** A body that no converter knows: a Kamelet route that keeps it gives
the same instance. */
+ private static final class Order {
+ }
+}
diff --git
a/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceBodyTypesTest.java
b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceBodyTypesTest.java
new file mode 100644
index 000000000000..50b8c53cee27
--- /dev/null
+++
b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceBodyTypesTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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.microprofile.faulttolerance;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.CircuitBreakerDefinition;
+import org.apache.camel.model.OnFallbackDefinition;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
+import org.apache.camel.spi.CircuitBreakerConstants;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import
org.eclipse.microprofile.faulttolerance.exceptions.CircuitBreakerOpenException;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The body types of the circuitBreaker and onFallback EIPs (CAMEL-25928),
checked at runtime with MicroProfile Fault
+ * Tolerance: the steps start with the body the circuit breaker got, and the
body after it is what its steps end with,
+ * or what onFallback ends with when a step failed or the circuit is open. The
EIPs implemented in core are checked by
+ * EipBodyTypesTest in camel-core.
+ */
+public class FaultToleranceBodyTypesTest extends CamelTestSupport {
+
+ /** The body the steps of circuitBreaker end with. */
+ private static final String END = "end of the steps";
+ /** The body the steps of onFallback end with. */
+ private static final String FALLBACK_END = "end of onFallback";
+ /** The body a step sets before the next step fails. */
+ private static final String FAILED = "the body the failed step left";
+
+ private final List<Object> steps = new CopyOnWriteArrayList<>();
+ private final List<Object> fallback = new CopyOnWriteArrayList<>();
+
+ @Test
+ public void theDeclarations() {
+ assertEquals("steps",
bodyTypes(CircuitBreakerDefinition.class).eipOut().from());
+ assertEquals(Carrier.UNCHANGED,
bodyTypes(CircuitBreakerDefinition.class).eipSteps().value()[0]);
+ assertEquals("steps",
bodyTypes(OnFallbackDefinition.class).eipOut().from());
+ assertEquals(Carrier.UNCHANGED,
bodyTypes(OnFallbackDefinition.class).eipSteps().value()[0]);
+ }
+
+ @Test
+ public void theStepsStartWithTheBodyAndTheBodyAfterIsWhatTheyEndWith() {
+ Order body = new Order();
+ Exchange out = send("direct:steps", body);
+ assertFalse(out.isFailed());
+ assertEquals(END, out.getMessage().getBody());
+ assertEquals(1, steps.size());
+ assertSame(body, steps.get(0));
+ assertTrue(fallback.isEmpty());
+ }
+
+ @Test
+ public void
whenAStepFailsOnFallbackStartsWithTheBodyTheCircuitBreakerGot() {
+ Order body = new Order();
+ Exchange out = send("direct:failed", body);
+ assertFalse(out.isFailed());
+ assertEquals(FALLBACK_END, out.getMessage().getBody());
+ assertEquals(1, steps.size());
+ assertSame(body, steps.get(0));
+ // the steps run on a copy of the exchange, so onFallback gets the
body the circuit breaker got,
+ // not the body the failed step left
+ assertEquals(1, fallback.size());
+ assertSame(body, fallback.get(0));
+ assertEquals(false,
out.getProperty(CircuitBreakerConstants.RESPONSE_REJECTED));
+ }
+
+ @Test
+ public void
whenTheCircuitIsOpenOnFallbackStartsWithTheBodyTheCircuitBreakerGot() {
+ // two failures open the circuit, the third call is not attempted
+ send("direct:open", new Order());
+ send("direct:open", new Order());
+
+ Order body = new Order();
+ Exchange out = send("direct:open", body);
+ assertFalse(out.isFailed());
+ assertEquals(FALLBACK_END, out.getMessage().getBody());
+ assertTrue(steps.isEmpty());
+ assertEquals(1, fallback.size());
+ assertSame(body, fallback.get(0));
+ assertEquals(true,
out.getProperty(CircuitBreakerConstants.RESPONSE_REJECTED));
+ assertInstanceOf(CircuitBreakerOpenException.class,
out.getProperty(Exchange.EXCEPTION_CAUGHT));
+ }
+
+ private Exchange send(String uri, Object body) {
+ steps.clear();
+ fallback.clear();
+ return template.send(uri, e -> e.getMessage().setBody(body));
+ }
+
+ private void steps(Exchange exchange) {
+ steps.add(exchange.getMessage().getBody());
+ }
+
+ private void fallback(Exchange exchange) {
+ fallback.add(exchange.getMessage().getBody());
+ }
+
+ private static BodyTypes bodyTypes(Class<?> model) {
+ BodyTypes answer = model.getAnnotation(BodyTypes.class);
+ assertNotNull(answer, model.getSimpleName() + " has no @BodyTypes");
+ return answer;
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:steps")
+ .circuitBreaker()
+ .process(e -> steps(e)).setBody(constant(END))
+ .onFallback()
+ .process(e ->
fallback(e)).setBody(constant(FALLBACK_END))
+ .end();
+
+ from("direct:failed")
+ .circuitBreaker()
+ .process(e -> steps(e)).setBody(constant(FAILED))
+ .throwException(new IllegalStateException("Forced"))
+ .onFallback()
+ .process(e ->
fallback(e)).setBody(constant(FALLBACK_END))
+ .end();
+
+ from("direct:open")
+ .circuitBreaker()
+
.faultToleranceConfiguration().requestVolumeThreshold(2).failureRatio(50).delay(60000).end()
+ .process(e -> steps(e)).setBody(constant(FAILED))
+ .throwException(new IllegalStateException("Forced"))
+ .onFallback()
+ .process(e ->
fallback(e)).setBody(constant(FALLBACK_END))
+ .end();
+ }
+ };
+ }
+
+ /** A body that no converter knows: a circuit breaker that keeps it gives
the same instance. */
+ private static final class Order {
+ }
+}
diff --git
a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/CircuitBreakerBodyTypesTest.java
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/CircuitBreakerBodyTypesTest.java
new file mode 100644
index 000000000000..0c3b0eb99972
--- /dev/null
+++
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/CircuitBreakerBodyTypesTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.resilience4j;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.CircuitBreakerDefinition;
+import org.apache.camel.model.OnFallbackDefinition;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
+import org.apache.camel.spi.CircuitBreakerConstants;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The body types of the circuitBreaker and onFallback EIPs (CAMEL-25928),
checked at runtime with Resilience4j: the
+ * steps start with the body the circuit breaker got, and the body after it is
what its steps end with, or what
+ * onFallback ends with when a step failed or the circuit is open. The EIPs
implemented in core are checked by
+ * EipBodyTypesTest in camel-core.
+ */
+public class CircuitBreakerBodyTypesTest extends CamelTestSupport {
+
+ /** The body the steps of circuitBreaker end with. */
+ private static final String END = "end of the steps";
+ /** The body the steps of onFallback end with. */
+ private static final String FALLBACK_END = "end of onFallback";
+ /** The body a step sets before the next step fails. */
+ private static final String FAILED = "the body the failed step left";
+
+ private final List<Object> steps = new CopyOnWriteArrayList<>();
+ private final List<Object> fallback = new CopyOnWriteArrayList<>();
+
+ @BindToRegistry("openCircuitBreaker")
+ public CircuitBreaker openCircuitBreaker() {
+ return CircuitBreaker.ofDefaults("openCircuitBreaker");
+ }
+
+ @Test
+ public void theDeclarations() {
+ assertEquals("steps",
bodyTypes(CircuitBreakerDefinition.class).eipOut().from());
+ assertEquals(Carrier.UNCHANGED,
bodyTypes(CircuitBreakerDefinition.class).eipSteps().value()[0]);
+ assertEquals("steps",
bodyTypes(OnFallbackDefinition.class).eipOut().from());
+ assertEquals(Carrier.UNCHANGED,
bodyTypes(OnFallbackDefinition.class).eipSteps().value()[0]);
+ }
+
+ @Test
+ public void theStepsStartWithTheBodyAndTheBodyAfterIsWhatTheyEndWith() {
+ Order body = new Order();
+ Exchange out = send("direct:steps", body);
+ assertFalse(out.isFailed());
+ assertEquals(END, out.getMessage().getBody());
+ assertEquals(1, steps.size());
+ assertSame(body, steps.get(0));
+ assertTrue(fallback.isEmpty());
+ }
+
+ @Test
+ public void
whenAStepFailsOnFallbackStartsWithTheBodyTheCircuitBreakerGot() {
+ Order body = new Order();
+ Exchange out = send("direct:failed", body);
+ assertFalse(out.isFailed());
+ assertEquals(FALLBACK_END, out.getMessage().getBody());
+ assertEquals(1, steps.size());
+ assertSame(body, steps.get(0));
+ // the steps run on a copy of the exchange, so onFallback gets the
body the circuit breaker got,
+ // not the body the failed step left
+ assertEquals(1, fallback.size());
+ assertSame(body, fallback.get(0));
+ assertEquals(false,
out.getProperty(CircuitBreakerConstants.RESPONSE_REJECTED));
+ }
+
+ @Test
+ public void
whenTheCircuitIsOpenOnFallbackStartsWithTheBodyTheCircuitBreakerGot() {
+ CircuitBreaker breaker =
context.getRegistry().lookupByNameAndType("openCircuitBreaker",
CircuitBreaker.class);
+ breaker.transitionToForcedOpenState();
+
+ Order body = new Order();
+ Exchange out = send("direct:open", body);
+ assertFalse(out.isFailed());
+ assertEquals(FALLBACK_END, out.getMessage().getBody());
+ // the call is not attempted
+ assertTrue(steps.isEmpty());
+ assertEquals(1, fallback.size());
+ assertSame(body, fallback.get(0));
+ assertEquals(true,
out.getProperty(CircuitBreakerConstants.RESPONSE_REJECTED));
+ assertInstanceOf(CallNotPermittedException.class,
out.getProperty(Exchange.EXCEPTION_CAUGHT));
+ }
+
+ private Exchange send(String uri, Object body) {
+ steps.clear();
+ fallback.clear();
+ return template.send(uri, e -> e.getMessage().setBody(body));
+ }
+
+ private void steps(Exchange exchange) {
+ steps.add(exchange.getMessage().getBody());
+ }
+
+ private void fallback(Exchange exchange) {
+ fallback.add(exchange.getMessage().getBody());
+ }
+
+ private static BodyTypes bodyTypes(Class<?> model) {
+ BodyTypes answer = model.getAnnotation(BodyTypes.class);
+ assertNotNull(answer, model.getSimpleName() + " has no @BodyTypes");
+ return answer;
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:steps")
+ .circuitBreaker()
+ .process(e -> steps(e)).setBody(constant(END))
+ .onFallback()
+ .process(e ->
fallback(e)).setBody(constant(FALLBACK_END))
+ .end();
+
+ from("direct:failed")
+ .circuitBreaker()
+ .process(e -> steps(e)).setBody(constant(FAILED))
+ .throwException(new IllegalStateException("Forced"))
+ .onFallback()
+ .process(e ->
fallback(e)).setBody(constant(FALLBACK_END))
+ .end();
+
+ from("direct:open")
+
.circuitBreaker().resilience4jConfiguration().circuitBreaker("openCircuitBreaker").end()
+ .process(e -> steps(e)).setBody(constant(END))
+ .onFallback()
+ .process(e ->
fallback(e)).setBody(constant(FALLBACK_END))
+ .end();
+ }
+ };
+ }
+
+ /** A body that no converter knows: a circuit breaker that keeps it gives
the same instance. */
+ private static final class Order {
+ }
+}
