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 020181ed8a63 CAMEL-26158: camel-vertx - reply with a stream as a
Buffer, and with a failure when the reply has no codec (#28941)
020181ed8a63 is described below
commit 020181ed8a632187d2f0dab51a150c239e600aba
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:50:54 2026 +0100
CAMEL-26158: camel-vertx - reply with a stream as a Buffer, and with a
failure when the reply has no codec (#28941)
* 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.
* CAMEL-26158: camel-vertx - the producer sends a stream body as a Buffer
With stream caching on, a Buffer that a vertx consumer receives becomes a
stream cache, so a
route from one vertx endpoint to another failed in the producer with "No
message codec for type".
The producer now sends a StreamCache or InputStream body as a Buffer, as
the consumer does for
its reply; the conversion is shared in VertxBodyHelper. Other bodies are
sent as is.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/components/vertx.json | 6 +-
.../apache/camel/catalog/docs/vertx-component.adoc | 18 +++--
.../org/apache/camel/component/vertx/vertx.json | 6 +-
.../camel-vertx/src/main/docs/vertx-component.adoc | 18 +++--
.../camel/component/vertx/VertxBodyHelper.java | 47 +++++++++++
.../camel/component/vertx/VertxConsumer.java | 12 ++-
.../camel/component/vertx/VertxEndpoint.java | 16 ++--
.../camel/component/vertx/VertxProducer.java | 7 +-
.../vertx/VertxRequestReplyFailureTest.java | 31 +++++++-
.../vertx/VertxRequestReplyStreamTest.java | 90 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +++
.../endpoint/dsl/VertxEndpointBuilderFactory.java | 24 ++++--
12 files changed, 254 insertions(+), 33 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..68330442fedd 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" },
- "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" },
+ "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", "stream", "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, such as a
received Buffer the route did not change, is sent as a Buffer. 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..930844f5e553 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,14 @@ A message goes to one consumer, or to all consumers with
`pubSub`:
message: "Received order ${body}"
----
-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.
+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, except for a stream, which is
sent as a `Buffer`: a `Buffer` the
+consumer receives is a stream when stream caching is enabled (the default), so
a route from one `vertx` endpoint to
+another sends it on as a `Buffer`. 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 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 +106,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..68330442fedd 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" },
- "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" },
+ "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", "stream", "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, such as a
received Buffer the route did not change, is sent as a Buffer. 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..930844f5e553 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,14 @@ A message goes to one consumer, or to all consumers with
`pubSub`:
message: "Received order ${body}"
----
-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.
+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, except for a stream, which is
sent as a `Buffer`: a `Buffer` the
+consumer receives is a stream when stream caching is enabled (the default), so
a route from one `vertx` endpoint to
+another sends it on as a `Buffer`. 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 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 +106,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/VertxBodyHelper.java
b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxBodyHelper.java
new file mode 100644
index 000000000000..bb4b723256b1
--- /dev/null
+++
b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxBodyHelper.java
@@ -0,0 +1,47 @@
+/*
+ * 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.InputStream;
+
+import io.vertx.core.buffer.Buffer;
+import org.apache.camel.Exchange;
+import org.apache.camel.StreamCache;
+import org.apache.camel.support.MessageHelper;
+
+final class VertxBodyHelper {
+
+ private VertxBodyHelper() {
+ }
+
+ /**
+ * The body to send on the event bus. A stream has no codec on the event
bus, so it is sent as a {@link Buffer}: a
+ * received {@link Buffer} is a stream when stream caching is enabled, and
a route that does not change it sends it
+ * on. Any other body is sent as is.
+ */
+ static Object eventBusBody(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;
+ }
+}
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..bac3f2ac5012 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
@@ -72,9 +72,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 =
VertxBodyHelper.eventBusBody(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);
+ }
}
}
}
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..7f2dd3161ae9 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,16 +45,19 @@ 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 },
+ Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, 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"
+ " 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 object such
as a HashMap. A stream, such as a received"
+ + " Buffer the route did not
change, is sent as a Buffer. A POJO that is not"
+ " Serializable (without a
codec registered for it) or no body fails:"
+ " convert it first"),
producerOut = @BodyType(value = Carrier.UNKNOWN,
@@ -143,7 +146,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..f881ec23a7e0 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
@@ -56,7 +56,7 @@ public class VertxProducer extends DefaultAsyncProducer {
boolean reply = ExchangeHelper.isOutCapable(exchange);
boolean pubSub = getEndpoint().isPubSub();
- Object body = exchange.getMessage().getBody();
+ Object body = VertxBodyHelper.eventBusBody(exchange);
if (body != null) {
if (reply) {
LOG.debug("Sending to: {} with body: {}", address, body);
@@ -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..47256e1d8708
--- /dev/null
+++
b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxRequestReplyStreamTest.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.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, and the producer sends
+ * it 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());
+ }
+
+ @Test
+ public void testForwardBuffer() {
+ // the route receives the Buffer as a stream cache and sends it on to
another vertx endpoint
+ Exchange out = template.request("vertx:foo.forward", e ->
e.getIn().setBody(Buffer.buffer("Camel")));
+
+ assertNull(out.getException());
+ assertEquals("Bye Camel", out.getMessage().getBody(String.class));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("vertx:foo.echo").to("mock:echo");
+
+ from("vertx:foo.forward").to("vertx:foo.bye");
+ from("vertx:foo.bye")
+ .transform(simple("Bye ${bodyAs(String)}"));
+
+ 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 eddc889d1481..5c3eff78d434 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
@@ -2486,6 +2486,18 @@ 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 - a stream body is sent as a Buffer, and the consumer replies
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).
+
+The `vertx` producer now sends a stream body as a `Buffer` too, so a route
from one `vertx` endpoint to another works
+with stream caching enabled. Before, it failed with `No message codec for
type`.
+
=== 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.
