This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26507-vertx-null-body in repository https://gitbox.apache.org/repos/asf/camel.git
commit 1d1887a3953bfa87729a9aaf5cef494f49838890 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 22:25:47 2026 +0100 CAMEL-26507: camel-vertx - the producer sends a message without a body as a null body The producer failed an exchange without a body with an InvalidPayloadRuntimeException, though the Vert.x event bus sends a null body with its built-in null codec, and a message without a body is a normal trigger or ping signal. Send, publish and request now send the null body; the consumer receives it as a message without a body and can reply to it. The body types and the component doc 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 | 3 +- .../org/apache/camel/component/vertx/vertx.json | 4 +- .../camel-vertx/src/main/docs/vertx-component.adoc | 3 +- .../camel/component/vertx/VertxEndpoint.java | 10 +- .../camel/component/vertx/VertxProducer.java | 35 ++---- .../camel/component/vertx/VertxNullBodyTest.java | 140 +++++++++++++++++++++ .../vertx/VertxRequestReplyFailureTest.java | 13 -- 8 files changed, 167 insertions(+), 45 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 68330442fedd..eb5e1bf8900a 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 @@ -35,9 +35,9 @@ "vertxFactory": { "index": 8, "kind": "property", "displayName": "Vertx Factory", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "io.vertx.core.VertxBuilder", "deprecated": false, "autowired": false, "secret": false, "description": "To use a custom VertxFactory implementation" } }, "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" }, + "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. A message sent without a body has a null body" }, "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" }, + "producerIn": { "carriers": [ "text", "bytes", "stream", "value", "map", "list", "pojo", "null" ], "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) fails: convert it first. A message without [...] "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": { 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 930844f5e553..75c4276e69f4 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 @@ -81,7 +81,8 @@ The body is sent as is, without type conversion, except for a stream, which is s 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`. +is sent with a `null` body, which the event bus supports with its built-in null codec, so it can be used as a trigger or +ping signal: the consumer receives a message without a body. This applies to send, publish and request alike. When the exchange is InOut, the producer sends a request and the reply of the consumer becomes the message body: 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 68330442fedd..eb5e1bf8900a 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 @@ -35,9 +35,9 @@ "vertxFactory": { "index": 8, "kind": "property", "displayName": "Vertx Factory", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "io.vertx.core.VertxBuilder", "deprecated": false, "autowired": false, "secret": false, "description": "To use a custom VertxFactory implementation" } }, "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" }, + "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. A message sent without a body has a null body" }, "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" }, + "producerIn": { "carriers": [ "text", "bytes", "stream", "value", "map", "list", "pojo", "null" ], "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) fails: convert it first. A message without [...] "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": { 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 930844f5e553..75c4276e69f4 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 @@ -81,7 +81,8 @@ The body is sent as is, without type conversion, except for a stream, which is s 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`. +is sent with a `null` body, which the event bus supports with its built-in null codec, so it can be used as a trigger or +ping signal: the consumer receives a message without a body. This applies to send, publish and request alike. When the exchange is InOut, the producer sends a request and the reply of the consumer becomes the message body: 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 7f2dd3161ae9..d888ea4d5d02 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 @@ -43,7 +43,8 @@ import org.apache.camel.support.DefaultEndpoint; 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"), + + " stream caching enabled. A message sent without a body has a null" + + " body"), consumerReply = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO, Carrier.NULL }, @@ -53,13 +54,14 @@ import org.apache.camel.support.DefaultEndpoint; + " Serializable (without a codec registered for it) is answered with a" + " failure"), producerIn = @BodyType(value = { - Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO }, + Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO, + Carrier.NULL }, 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"), + + " Serializable (without a codec registered for it) fails: convert it" + + " first. A message without a body is sent with a null body"), producerOut = @BodyType(value = Carrier.UNKNOWN, byOption = @BodyType.When(option = "exchangePattern", value = "InOnly", carriers = Carrier.UNCHANGED), 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 f881ec23a7e0..954f24779fc9 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 @@ -22,7 +22,6 @@ import io.vertx.core.eventbus.EventBus; import io.vertx.core.eventbus.Message; import org.apache.camel.AsyncCallback; import org.apache.camel.Exchange; -import org.apache.camel.InvalidPayloadRuntimeException; import org.apache.camel.support.DefaultAsyncProducer; import org.apache.camel.support.ExchangeHelper; import org.apache.camel.support.MessageHelper; @@ -56,31 +55,23 @@ public class VertxProducer extends DefaultAsyncProducer { boolean reply = ExchangeHelper.isOutCapable(exchange); boolean pubSub = getEndpoint().isPubSub(); + // a message without a body is sent with a null body, which the event bus sends with its null codec Object body = VertxBodyHelper.eventBusBody(exchange); - if (body != null) { - if (reply) { - LOG.debug("Sending to: {} with body: {}", address, body); - eventBus.<Object> request(address, body).onComplete(new CamelReplyHandler(exchange, callback)); - return false; + if (reply) { + LOG.debug("Sending to: {} with body: {}", address, body); + eventBus.<Object> request(address, body).onComplete(new CamelReplyHandler(exchange, callback)); + return false; + } else { + if (pubSub) { + LOG.debug("Publishing to: {} with body: {}", address, body); + eventBus.publish(address, body); } else { - if (pubSub) { - LOG.debug("Publishing to: {} with body: {}", address, body); - eventBus.publish(address, body); - } else { - LOG.debug("Sending to: {} with body: {}", address, body); - eventBus.send(address, body); - } - callback.done(true); - return true; + LOG.debug("Sending to: {} with body: {}", address, body); + eventBus.send(address, body); } + callback.done(true); + return true; } - - // 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; } private static final class CamelReplyHandler implements Handler<AsyncResult<Message<Object>>> { diff --git a/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxNullBodyTest.java b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxNullBodyTest.java new file mode 100644 index 000000000000..d07f72312def --- /dev/null +++ b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxNullBodyTest.java @@ -0,0 +1,140 @@ +/* + * 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.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +import io.vertx.core.eventbus.EventBus; +import io.vertx.core.eventbus.Message; +import io.vertx.core.eventbus.MessageConsumer; +import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; +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.assertNull; + +/** + * A message without a body is sent to the event bus with a null body (the event bus has a codec for null), with send, + * publish and request, and a consumer receives it. + */ +public class VertxNullBodyTest extends VertxBaseTestSupport { + + @Test + public void testSendNullBody() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:send"); + mock.expectedMessageCount(1); + mock.message(0).body().isNull(); + + template.sendBody("vertx:foo.send", null); + + mock.assertIsSatisfied(); + } + + @Test + public void testPublishNullBody() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:publish"); + // both consumers of the address receive it + mock.expectedMessageCount(2); + mock.allMessages().body().isNull(); + + template.sendBody("vertx:foo.publish?pubSub=true", null); + + mock.assertIsSatisfied(); + } + + @Test + public void testRequestNullBody() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:request"); + mock.expectedMessageCount(1); + mock.message(0).body().isNull(); + + Object reply = template.requestBody("vertx:foo.request", (Object) null); + + mock.assertIsSatisfied(); + assertEquals("Pong", reply); + } + + @Test + public void testRequestNullBodyWithNullReply() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:echo"); + mock.expectedMessageCount(1); + mock.message(0).body().isNull(); + + Exchange out = template.request("vertx:foo.echo", e -> e.getIn().setBody(null)); + + mock.assertIsSatisfied(); + assertNull(out.getException()); + assertNull(out.getMessage().getBody()); + } + + @Test + public void testEventBusHandlerReceivesNullBody() throws Exception { + EventBus eventBus = context.getComponent("vertx", VertxComponent.class).getVertx().eventBus(); + CompletableFuture<Message<Object>> received = new CompletableFuture<>(); + MessageConsumer<Object> consumer = eventBus.consumer("foo.handler", received::complete); + consumer.completion().toCompletionStage().toCompletableFuture().get(10, TimeUnit.SECONDS); + try { + template.sendBody("vertx:foo.handler", null); + + Message<Object> message = received.get(10, TimeUnit.SECONDS); + assertNull(message.body()); + assertNull(message.replyAddress()); + } finally { + consumer.unregister(); + } + } + + @Test + public void testEventBusHandlerRepliesToNullBodyRequest() throws Exception { + EventBus eventBus = context.getComponent("vertx", VertxComponent.class).getVertx().eventBus(); + CompletableFuture<Object> received = new CompletableFuture<>(); + MessageConsumer<Object> consumer = eventBus.consumer("foo.handler.request", message -> { + received.complete(message.body()); + message.reply(message.body() == null ? "Pong" : "Unexpected"); + }); + consumer.completion().toCompletionStage().toCompletableFuture().get(10, TimeUnit.SECONDS); + try { + Object reply = template.sendBody("vertx:foo.handler.request", ExchangePattern.InOut, null); + + assertNull(received.get(10, TimeUnit.SECONDS)); + assertEquals("Pong", reply); + } finally { + consumer.unregister(); + } + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("vertx:foo.send").to("mock:send"); + + from("vertx:foo.publish").to("mock:publish"); + from("vertx:foo.publish").to("mock:publish"); + + from("vertx:foo.request").to("mock:request").setBody(constant("Pong")); + + // replies with the body it received, which is null + from("vertx:foo.echo").to("mock:echo"); + } + }; + } +} 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 f337c6578639..d08f6ad092a7 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,14 +19,12 @@ 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 @@ -64,15 +62,6 @@ public class VertxRequestReplyFailureTest extends VertxBaseTestSupport { 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() { @@ -89,8 +78,6 @@ public class VertxRequestReplyFailureTest extends VertxBaseTestSupport { 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"); } }; }
