This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26158-vertx-reply-codec in repository https://gitbox.apache.org/repos/asf/camel.git
commit d901940f54c4641f15c8eb852d48437d1da76080 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 18:00:51 2026 +0100 CAMEL-26158: camel-vertx - reply with a stream as a Buffer, and with a failure when the reply has no codec The consumer called event.reply(body) and nobody caught the IllegalArgumentException that Vert.x throws when the body has no codec, so no reply was sent and the requester waited until its timeout. An echo route of a Buffer request hit this too, as stream caching turns the Buffer into a stream cache. The consumer now replies with a stream body as a Buffer, and when the reply still cannot be sent it sends a failure reply and passes the error to the exception handler. The producer failed a message without a body with "No body available of type: java.lang.String", though it sends any body type; the error now says that it cannot send a message without a body. The pubSub option only applies to InOnly exchanges; its description and the docs now say so. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/components/vertx.json | 4 +- .../apache/camel/catalog/docs/vertx-component.adoc | 12 +++- .../org/apache/camel/component/vertx/vertx.json | 4 +- .../camel-vertx/src/main/docs/vertx-component.adoc | 12 +++- .../camel/component/vertx/VertxConsumer.java | 33 +++++++++- .../camel/component/vertx/VertxEndpoint.java | 11 ++-- .../camel/component/vertx/VertxProducer.java | 5 +- .../vertx/VertxRequestReplyFailureTest.java | 31 ++++++++- .../vertx/VertxRequestReplyStreamTest.java | 76 ++++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +++ .../endpoint/dsl/VertxEndpointBuilderFactory.java | 24 +++++-- 11 files changed, 197 insertions(+), 24 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/vertx.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/vertx.json index 33549d87d679..34b7880e44f4 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/vertx.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/vertx.json @@ -36,13 +36,13 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "unknown" ], "description": "The body of the event bus message, as the sender sent it: a String, a byte array, a number or a boolean, a JsonObject or a JsonArray, or a Serializable object such as a HashMap. A Buffer becomes a stream with stream caching enabled" }, - "consumerReply": { "carriers": [ "text", "bytes", "value", "map", "list", "pojo", "null" ], "description": "A message with a reply address is answered with the body the route ends with, as the producer sends it. A stream, such as a received Buffer the route did not change, or a POJO that is not Serializable sends no reply, and the requester times out" }, + "consumerReply": { "carriers": [ "text", "bytes", "stream", "value", "map", "list", "pojo", "null" ], "description": "A message with a reply address is answered with the body the route ends with, as the producer sends it. A stream, such as a received Buffer the route did not change, is replied as a Buffer. A POJO that is not Serializable (without a codec registered for it) is answered with a failure" }, "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list", "pojo" ], "description": "Sent as is, with the event bus codec of its type: a String, a byte array, a number or a boolean, a JsonObject, a JsonArray or a Buffer, or a Serializable object such as a HashMap. A stream, a POJO that is not Serializable (without a codec registered for it) or no body fails: convert it first" }, "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option": "exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] } ], "description": "Unchanged for a message that expects no reply (InOnly). A message that expects one (InOut) is sent as a request, also with pubSub, and gets the body of the reply" } }, "properties": { "address": { "index": 0, "kind": "path", "displayName": "Address", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Sets the event bus address used to communicate" }, - "pubSub": { "index": 1, "kind": "parameter", "displayName": "Pub Sub", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to use publish\/subscribe instead of point to point when sending to a vertx endpoint." }, + "pubSub": { "index": 1, "kind": "parameter", "displayName": "Pub Sub", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to use publish\/subscribe instead of point to point when sending to a vertx endpoint. This only applies when the exchange pattern is InOnly: an InOut exchange is sent as a request to one consumer, which replies." }, "bridgeErrorHandler": { "index": 2, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming [...] "exceptionHandler": { "index": 3, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By def [...] "exchangePattern": { "index": 4, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/vertx-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/vertx-component.adoc index d097b8eb6a6b..2ebc2a8fca05 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/vertx-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/vertx-component.adoc @@ -74,9 +74,12 @@ A message goes to one consumer, or to all consumers with `pubSub`: message: "Received order ${body}" ---- +The `pubSub` option only applies to InOnly exchanges: an InOut exchange is always sent as a request to one consumer, +which replies. + The body is sent as is, without type conversion: a stream, or a POJO that is not `Serializable` and has no codec registered on the event bus, fails with `No message codec for type`, so convert it to a `String` or a byte array -first. A message without a body fails as well. +first. A message without a body fails with an `InvalidPayloadRuntimeException`. When the exchange is InOut, the producer sends a request and the reply of the consumer becomes the message body: @@ -101,8 +104,11 @@ When the exchange is InOut, the producer sends a request and the reply of the co expression: "Hello ${body}" ---- -The consumer replies with the body the route ends with, which needs a codec too. A `Buffer` the consumer receives is -a stream when stream caching is enabled (the default), so convert a stream body before the route ends. +The consumer replies with the body the route ends with. A stream body is replied as a `Buffer`: a `Buffer` the +consumer receives is a stream when stream caching is enabled (the default), so a route that does not change the body +replies with a `Buffer` again. Any other body needs a codec too: when the reply cannot be sent, such as for a POJO +that is not `Serializable`, the consumer logs the error and the sender gets a failure reply (a `ReplyException` with +failure code 500) instead of waiting until its reply timeout. == Connecting to the existing Vert.x instance diff --git a/components/camel-vertx/camel-vertx/src/generated/resources/META-INF/org/apache/camel/component/vertx/vertx.json b/components/camel-vertx/camel-vertx/src/generated/resources/META-INF/org/apache/camel/component/vertx/vertx.json index 33549d87d679..34b7880e44f4 100644 --- a/components/camel-vertx/camel-vertx/src/generated/resources/META-INF/org/apache/camel/component/vertx/vertx.json +++ b/components/camel-vertx/camel-vertx/src/generated/resources/META-INF/org/apache/camel/component/vertx/vertx.json @@ -36,13 +36,13 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "unknown" ], "description": "The body of the event bus message, as the sender sent it: a String, a byte array, a number or a boolean, a JsonObject or a JsonArray, or a Serializable object such as a HashMap. A Buffer becomes a stream with stream caching enabled" }, - "consumerReply": { "carriers": [ "text", "bytes", "value", "map", "list", "pojo", "null" ], "description": "A message with a reply address is answered with the body the route ends with, as the producer sends it. A stream, such as a received Buffer the route did not change, or a POJO that is not Serializable sends no reply, and the requester times out" }, + "consumerReply": { "carriers": [ "text", "bytes", "stream", "value", "map", "list", "pojo", "null" ], "description": "A message with a reply address is answered with the body the route ends with, as the producer sends it. A stream, such as a received Buffer the route did not change, is replied as a Buffer. A POJO that is not Serializable (without a codec registered for it) is answered with a failure" }, "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list", "pojo" ], "description": "Sent as is, with the event bus codec of its type: a String, a byte array, a number or a boolean, a JsonObject, a JsonArray or a Buffer, or a Serializable object such as a HashMap. A stream, a POJO that is not Serializable (without a codec registered for it) or no body fails: convert it first" }, "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option": "exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] } ], "description": "Unchanged for a message that expects no reply (InOnly). A message that expects one (InOut) is sent as a request, also with pubSub, and gets the body of the reply" } }, "properties": { "address": { "index": 0, "kind": "path", "displayName": "Address", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Sets the event bus address used to communicate" }, - "pubSub": { "index": 1, "kind": "parameter", "displayName": "Pub Sub", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to use publish\/subscribe instead of point to point when sending to a vertx endpoint." }, + "pubSub": { "index": 1, "kind": "parameter", "displayName": "Pub Sub", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to use publish\/subscribe instead of point to point when sending to a vertx endpoint. This only applies when the exchange pattern is InOnly: an InOut exchange is sent as a request to one consumer, which replies." }, "bridgeErrorHandler": { "index": 2, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming [...] "exceptionHandler": { "index": 3, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By def [...] "exchangePattern": { "index": 4, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, diff --git a/components/camel-vertx/camel-vertx/src/main/docs/vertx-component.adoc b/components/camel-vertx/camel-vertx/src/main/docs/vertx-component.adoc index d097b8eb6a6b..2ebc2a8fca05 100644 --- a/components/camel-vertx/camel-vertx/src/main/docs/vertx-component.adoc +++ b/components/camel-vertx/camel-vertx/src/main/docs/vertx-component.adoc @@ -74,9 +74,12 @@ A message goes to one consumer, or to all consumers with `pubSub`: message: "Received order ${body}" ---- +The `pubSub` option only applies to InOnly exchanges: an InOut exchange is always sent as a request to one consumer, +which replies. + The body is sent as is, without type conversion: a stream, or a POJO that is not `Serializable` and has no codec registered on the event bus, fails with `No message codec for type`, so convert it to a `String` or a byte array -first. A message without a body fails as well. +first. A message without a body fails with an `InvalidPayloadRuntimeException`. When the exchange is InOut, the producer sends a request and the reply of the consumer becomes the message body: @@ -101,8 +104,11 @@ When the exchange is InOut, the producer sends a request and the reply of the co expression: "Hello ${body}" ---- -The consumer replies with the body the route ends with, which needs a codec too. A `Buffer` the consumer receives is -a stream when stream caching is enabled (the default), so convert a stream body before the route ends. +The consumer replies with the body the route ends with. A stream body is replied as a `Buffer`: a `Buffer` the +consumer receives is a stream when stream caching is enabled (the default), so a route that does not change the body +replies with a `Buffer` again. Any other body needs a codec too: when the reply cannot be sent, such as for a POJO +that is not `Serializable`, the consumer logs the error and the sender gets a failure reply (a `ReplyException` with +failure code 500) instead of waiting until its reply timeout. == Connecting to the existing Vert.x instance diff --git a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxConsumer.java b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxConsumer.java index d66e9429193e..f8edaad3da70 100644 --- a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxConsumer.java +++ b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxConsumer.java @@ -16,14 +16,19 @@ */ package org.apache.camel.component.vertx; +import java.io.InputStream; + import io.vertx.core.Handler; +import io.vertx.core.buffer.Buffer; import io.vertx.core.eventbus.Message; import io.vertx.core.eventbus.MessageConsumer; import org.apache.camel.AsyncCallback; import org.apache.camel.Exchange; import org.apache.camel.ExchangePattern; import org.apache.camel.Processor; +import org.apache.camel.StreamCache; import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.support.MessageHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,9 +77,17 @@ public class VertxConsumer extends DefaultConsumer { event.fail(FAILURE_CODE, FAILURE_MESSAGE); } else { // always reply (also with no body) so the sender does not wait until its timeout - Object body = exchange.getMessage().getBody(); + Object body = replyBody(exchange); LOG.debug("Sending reply to: {} with body: {}", event.replyAddress(), body); - event.reply(body); + try { + event.reply(body); + } catch (Exception e) { + // the body has no codec on the event bus, such as a POJO that is not Serializable, + // so let the sender know that the request failed instead of waiting until its timeout + event.fail(FAILURE_CODE, FAILURE_MESSAGE); + getExceptionHandler().handleException( + "Error sending reply to: " + event.replyAddress(), exchange, e); + } } } } @@ -84,6 +97,22 @@ public class VertxConsumer extends DefaultConsumer { } } + /** + * The body to reply with. A stream has no codec on the event bus, so it is replied as a {@link Buffer}: a received + * {@link Buffer} is a stream when stream caching is enabled, and a route that does not change it replies with it. + */ + private static Object replyBody(Exchange exchange) { + Object body = exchange.getMessage().getBody(); + if (body instanceof StreamCache || body instanceof InputStream) { + MessageHelper.resetStreamCache(exchange.getMessage()); + byte[] bytes = exchange.getContext().getTypeConverter().convertTo(byte[].class, exchange, body); + if (bytes != null) { + return Buffer.buffer(bytes); + } + } + return body; + } + @Override protected void doStart() throws Exception { if (LOG.isDebugEnabled()) { diff --git a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java index 2ccba990ee1c..38eb93c39f52 100644 --- a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java +++ b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java @@ -45,11 +45,13 @@ import org.apache.camel.support.DefaultEndpoint; + " Serializable object such as a HashMap. A Buffer becomes a stream with" + " stream caching enabled"), consumerReply = @BodyType(value = { - Carrier.TEXT, Carrier.BYTES, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO, Carrier.NULL }, + Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO, + Carrier.NULL }, description = "A message with a reply address is answered with the body the route ends" + " with, as the producer sends it. A stream, such as a received Buffer the" - + " route did not change, or a POJO that is not Serializable sends no reply," - + " and the requester times out"), + + " route did not change, is replied as a Buffer. A POJO that is not" + + " Serializable (without a codec registered for it) is answered with a" + + " failure"), producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO }, description = "Sent as is, with the event bus codec of its type: a String, a byte array, a" @@ -143,7 +145,8 @@ public class VertxEndpoint extends DefaultEndpoint implements AsyncEndpoint, Mul } /** - * Whether to use publish/subscribe instead of point to point when sending to a vertx endpoint. + * Whether to use publish/subscribe instead of point to point when sending to a vertx endpoint. This only applies + * when the exchange pattern is InOnly: an InOut exchange is sent as a request to one consumer, which replies. */ public void setPubSub(Boolean pubSub) { this.pubSub = pubSub; diff --git a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxProducer.java b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxProducer.java index a1b292626c81..fb358aabc3ca 100644 --- a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxProducer.java +++ b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxProducer.java @@ -75,7 +75,10 @@ public class VertxProducer extends DefaultAsyncProducer { } } - exchange.setException(new InvalidPayloadRuntimeException(exchange, String.class)); + // the producer sends any body type, but not a message without a body + exchange.setException(new InvalidPayloadRuntimeException( + exchange, Object.class, + new IllegalArgumentException("Cannot send a message without a body to the event bus address: " + address))); callback.done(true); return true; } diff --git a/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyFailureTest.java b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyFailureTest.java index 72327afbae70..f337c6578639 100644 --- a/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyFailureTest.java +++ b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyFailureTest.java @@ -19,16 +19,18 @@ package org.apache.camel.component.vertx; import io.vertx.core.eventbus.ReplyException; import io.vertx.core.eventbus.ReplyFailure; import org.apache.camel.Exchange; +import org.apache.camel.InvalidPayloadRuntimeException; import org.apache.camel.builder.RouteBuilder; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * The reply of a request/reply over the event bus must tell the sender that the route failed, and must also be sent - * when the route returns no body. + * when the route returns no body or a body that has no codec on the event bus. */ public class VertxRequestReplyFailureTest extends VertxBaseTestSupport { @@ -51,6 +53,26 @@ public class VertxRequestReplyFailureTest extends VertxBaseTestSupport { assertNull(out.getMessage().getBody()); } + @Test + public void testReplyWithoutCodecRepliesWithFailure() { + Exchange out = template.request("direct:nocodec", e -> e.getIn().setBody("Camel")); + + // a body that has no codec cannot be replied, so the sender gets a failure instead of waiting for its timeout + ReplyException cause = assertInstanceOf(ReplyException.class, out.getException()); + assertEquals(ReplyFailure.RECIPIENT_FAILURE, cause.failureType()); + assertEquals(VertxConsumer.FAILURE_CODE, cause.failureCode()); + assertEquals(VertxConsumer.FAILURE_MESSAGE, cause.getMessage()); + } + + @Test + public void testSendWithoutBodyFails() { + Exchange out = template.send("direct:nobody", e -> e.getIn().setBody(null)); + + InvalidPayloadRuntimeException cause = assertInstanceOf(InvalidPayloadRuntimeException.class, out.getException()); + assertTrue(cause.getMessage().contains("Cannot send a message without a body to the event bus address: foo.nobody"), + cause.getMessage()); + } + @Override protected RouteBuilder createRouteBuilder() { return new RouteBuilder() { @@ -62,6 +84,13 @@ public class VertxRequestReplyFailureTest extends VertxBaseTestSupport { from("direct:empty").to("vertx:foo.empty"); from("vertx:foo.empty") .setBody(constant(null)); + + from("direct:nocodec").to("vertx:foo.nocodec"); + from("vertx:foo.nocodec") + // an Object is not Serializable and has no codec on the event bus + .setBody(e -> new Object()); + + from("direct:nobody").to("vertx:foo.nobody"); } }; } diff --git a/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyStreamTest.java b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyStreamTest.java new file mode 100644 index 000000000000..9d06d917acd0 --- /dev/null +++ b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyStreamTest.java @@ -0,0 +1,76 @@ +/* + * 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.vertx; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; + +import io.vertx.core.buffer.Buffer; +import org.apache.camel.Exchange; +import org.apache.camel.StreamCache; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * A stream has no codec on the event bus, so the consumer replies with its content as a Buffer. + */ +public class VertxRequestReplyStreamTest extends VertxBaseTestSupport { + + @Test + public void testEchoBuffer() throws Exception { + // with stream caching enabled, the received Buffer is a stream cache that the route does not change + MockEndpoint mock = getMockEndpoint("mock:echo"); + mock.expectedMessageCount(1); + mock.message(0).body().isInstanceOf(StreamCache.class); + + // send from the template, as a route would turn the Buffer into a stream cache before the producer + Exchange out = template.request("vertx:foo.echo", e -> e.getIn().setBody(Buffer.buffer("Camel"))); + + MockEndpoint.assertIsSatisfied(context); + assertNull(out.getException()); + Buffer reply = assertInstanceOf(Buffer.class, out.getMessage().getBody()); + assertEquals("Camel", reply.toString()); + } + + @Test + public void testReplyInputStream() { + Exchange out = template.request("direct:stream", e -> e.getIn().setBody("Camel")); + + assertNull(out.getException()); + Buffer reply = assertInstanceOf(Buffer.class, out.getMessage().getBody()); + assertEquals("Hello Camel", reply.toString()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("vertx:foo.echo").to("mock:echo"); + + from("direct:stream").to("vertx:foo.stream"); + from("vertx:foo.stream") + .setBody(e -> new ByteArrayInputStream( + ("Hello " + e.getMessage().getBody(String.class)).getBytes(StandardCharsets.UTF_8))); + } + }; + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 1bc188419d61..729f3c0b543f 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -2246,6 +2246,15 @@ When configuring TLS through `SSLContextParameters` with a standalone `TrustMana When both a `TrustManager` and a keystore are configured on `TrustManagersParameters`, the `TrustManager` is now used, as with the other components that use `SSLContextParameters`; before, the keystore-based trust options were used. +=== camel-vertx - the consumer replies with a stream as a Buffer, and with a failure when the body has no codec + +When a `vertx` consumer receives a message that expects a reply and the route ends with a stream body (such as a +received `Buffer`, which is a stream with stream caching enabled, that the route did not change), the consumer now +replies with its content as a `Buffer`. When the reply body has no codec on the event bus (such as a POJO that is not +`Serializable`), the consumer now logs the error and sends a failure reply (a `ReplyException` with failure type +`RECIPIENT_FAILURE`, failure code 500 and the message `Exchange processing failed`). Before, both sent no reply, and +the sender waited until its reply timeout (30 seconds by default). + === camel-knative-http - a header with several values is sent as several headers A message header whose value is a `java.util.Collection` or an object array (as the consumer maps a request header diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/VertxEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/VertxEndpointBuilderFactory.java index 9c363a1904c1..8efd29af60af 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/VertxEndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/VertxEndpointBuilderFactory.java @@ -45,7 +45,9 @@ public interface VertxEndpointBuilderFactory { } /** * Whether to use publish/subscribe instead of point to point when - * sending to a vertx endpoint. + * sending to a vertx endpoint. This only applies when the exchange + * pattern is InOnly: an InOut exchange is sent as a request to one + * consumer, which replies. * * The option is a: <code>java.lang.Boolean</code> type. * @@ -61,7 +63,9 @@ public interface VertxEndpointBuilderFactory { } /** * Whether to use publish/subscribe instead of point to point when - * sending to a vertx endpoint. + * sending to a vertx endpoint. This only applies when the exchange + * pattern is InOnly: an InOut exchange is sent as a request to one + * consumer, which replies. * * The option will be converted to a <code>java.lang.Boolean</code> * type. @@ -218,7 +222,9 @@ public interface VertxEndpointBuilderFactory { /** * Whether to use publish/subscribe instead of point to point when - * sending to a vertx endpoint. + * sending to a vertx endpoint. This only applies when the exchange + * pattern is InOnly: an InOut exchange is sent as a request to one + * consumer, which replies. * * The option is a: <code>java.lang.Boolean</code> type. * @@ -234,7 +240,9 @@ public interface VertxEndpointBuilderFactory { } /** * Whether to use publish/subscribe instead of point to point when - * sending to a vertx endpoint. + * sending to a vertx endpoint. This only applies when the exchange + * pattern is InOnly: an InOut exchange is sent as a request to one + * consumer, which replies. * * The option will be converted to a <code>java.lang.Boolean</code> * type. @@ -320,7 +328,9 @@ public interface VertxEndpointBuilderFactory { /** * Whether to use publish/subscribe instead of point to point when - * sending to a vertx endpoint. + * sending to a vertx endpoint. This only applies when the exchange + * pattern is InOnly: an InOut exchange is sent as a request to one + * consumer, which replies. * * The option is a: <code>java.lang.Boolean</code> type. * @@ -336,7 +346,9 @@ public interface VertxEndpointBuilderFactory { } /** * Whether to use publish/subscribe instead of point to point when - * sending to a vertx endpoint. + * sending to a vertx endpoint. This only applies when the exchange + * pattern is InOnly: an InOut exchange is sent as a request to one + * consumer, which replies. * * The option will be converted to a <code>java.lang.Boolean</code> * type.
