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 d10c0c9d1174 CAMEL-26188: camel-cometd - publish a stream or file body
as its text, and a null body as a JSON null (#29014)
d10c0c9d1174 is described below
commit d10c0c9d1174db055b370c59e26c974823b3204f
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 08:56:54 2026 +0100
CAMEL-26188: camel-cometd - publish a stream or file body as its text, and
a null body as a JSON null (#29014)
* CAMEL-26188: camel-cometd - publish a stream or file body as its text…
…, and a null body as a JSON null
The producer and the reply of an InOut consumer set the body as is as the
Bayeux data, so a stream, a Reader
or a file reached the clients as the text of its toString. They are now
read as text before publishing.
A null body called setData(null), which leaves out the data field: the
clients took the message for the
reply of a publish and never delivered it to their subscribers. The data
field is now set to a JSON null.
* Merge remote-tracking branch 'origin/main' into CAMEL-26188-cometd-body
# Conflicts:
# docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
* CAMEL-26188: camel-cometd - a stream or file body that does not conve…
…rt to text fails instead of publishing no data
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/cometd.json | 6 +-
.../apache/camel/catalog/components/cometds.json | 6 +-
.../org/apache/camel/component/cometd/cometd.json | 6 +-
.../org/apache/camel/component/cometd/cometds.json | 6 +-
.../camel/component/cometd/CometdBinding.java | 30 ++-
.../camel/component/cometd/CometdEndpoint.java | 23 ++-
.../camel/component/cometd/CometdProducer.java | 4 +-
.../component/cometd/CometdProducerBodyTest.java | 201 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 +
9 files changed, 268 insertions(+), 24 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometd.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometd.json
index 5737e8a07f07..8dfb68350a75 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometd.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometd.json
@@ -41,9 +41,9 @@
"subscription": { "index": 1, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The subscription", "constantName":
"org.apache.camel.component.cometd.CometdBinding#COMETD_SUBSCRIPTION_HEADER_NAME"
}
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is" },
- "consumerReply": { "carriers": [ "text", "value", "map", "list" ],
"description": "Only when the exchange is InOut (exchangePattern=InOut, or set
in the route): the body the route ends with is sent back to the client that
published, as JSON. Bytes become an array of numbers; a stream, a file or
another object is sent as the text of its toString" },
- "producerIn": { "carriers": [ "text", "value", "map", "list" ],
"description": "The data published to the channel, sent as JSON to the
subscribed clients: text, a number or a boolean, a Map or a List. Bytes become
an array of numbers; a stream, a file or another object is sent as the text of
its toString: convert it first. The headers go in the ext field of the message"
},
+ "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is (a stream or
a file as its text)" },
+ "consumerReply": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "Only when the exchange is InOut
(exchangePattern=InOut, or set in the route): the body the route ends with is
sent back to the client that published, as JSON. A stream or a file is read as
text, and null is sent as a JSON null. Bytes become an array of numbers;
another object is sent as the text of its toString" },
+ "producerIn": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "The data published to the channel, sent as
JSON to the subscribed clients: text, a number or a boolean, a Map or a List. A
stream or a file is read as text, and null is published as a JSON null. Bytes
become an array of numbers; another object is sent as the text of its toString:
convert it first. The headers go in the ext field of the message" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometds.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometds.json
index f20d00b45f43..bb734eaa3df7 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometds.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cometds.json
@@ -41,9 +41,9 @@
"subscription": { "index": 1, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The subscription", "constantName":
"org.apache.camel.component.cometd.CometdBinding#COMETD_SUBSCRIPTION_HEADER_NAME"
}
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is" },
- "consumerReply": { "carriers": [ "text", "value", "map", "list" ],
"description": "Only when the exchange is InOut (exchangePattern=InOut, or set
in the route): the body the route ends with is sent back to the client that
published, as JSON. Bytes become an array of numbers; a stream, a file or
another object is sent as the text of its toString" },
- "producerIn": { "carriers": [ "text", "value", "map", "list" ],
"description": "The data published to the channel, sent as JSON to the
subscribed clients: text, a number or a boolean, a Map or a List. Bytes become
an array of numbers; a stream, a file or another object is sent as the text of
its toString: convert it first. The headers go in the ext field of the message"
},
+ "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is (a stream or
a file as its text)" },
+ "consumerReply": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "Only when the exchange is InOut
(exchangePattern=InOut, or set in the route): the body the route ends with is
sent back to the client that published, as JSON. A stream or a file is read as
text, and null is sent as a JSON null. Bytes become an array of numbers;
another object is sent as the text of its toString" },
+ "producerIn": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "The data published to the channel, sent as
JSON to the subscribed clients: text, a number or a boolean, a Map or a List. A
stream or a file is read as text, and null is published as a JSON null. Bytes
become an array of numbers; another object is sent as the text of its toString:
convert it first. The headers go in the ext field of the message" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometd.json
b/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometd.json
index 5737e8a07f07..8dfb68350a75 100644
---
a/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometd.json
+++
b/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometd.json
@@ -41,9 +41,9 @@
"subscription": { "index": 1, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The subscription", "constantName":
"org.apache.camel.component.cometd.CometdBinding#COMETD_SUBSCRIPTION_HEADER_NAME"
}
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is" },
- "consumerReply": { "carriers": [ "text", "value", "map", "list" ],
"description": "Only when the exchange is InOut (exchangePattern=InOut, or set
in the route): the body the route ends with is sent back to the client that
published, as JSON. Bytes become an array of numbers; a stream, a file or
another object is sent as the text of its toString" },
- "producerIn": { "carriers": [ "text", "value", "map", "list" ],
"description": "The data published to the channel, sent as JSON to the
subscribed clients: text, a number or a boolean, a Map or a List. Bytes become
an array of numbers; a stream, a file or another object is sent as the text of
its toString: convert it first. The headers go in the ext field of the message"
},
+ "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is (a stream or
a file as its text)" },
+ "consumerReply": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "Only when the exchange is InOut
(exchangePattern=InOut, or set in the route): the body the route ends with is
sent back to the client that published, as JSON. A stream or a file is read as
text, and null is sent as a JSON null. Bytes become an array of numbers;
another object is sent as the text of its toString" },
+ "producerIn": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "The data published to the channel, sent as
JSON to the subscribed clients: text, a number or a boolean, a Map or a List. A
stream or a file is read as text, and null is published as a JSON null. Bytes
become an array of numbers; another object is sent as the text of its toString:
convert it first. The headers go in the ext field of the message" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometds.json
b/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometds.json
index f20d00b45f43..bb734eaa3df7 100644
---
a/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometds.json
+++
b/components/camel-cometd/src/generated/resources/META-INF/org/apache/camel/component/cometd/cometds.json
@@ -41,9 +41,9 @@
"subscription": { "index": 1, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The subscription", "constantName":
"org.apache.camel.component.cometd.CometdBinding#COMETD_SUBSCRIPTION_HEADER_NAME"
}
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is" },
- "consumerReply": { "carriers": [ "text", "value", "map", "list" ],
"description": "Only when the exchange is InOut (exchangePattern=InOut, or set
in the route): the body the route ends with is sent back to the client that
published, as JSON. Bytes become an array of numbers; a stream, a file or
another object is sent as the text of its toString" },
- "producerIn": { "carriers": [ "text", "value", "map", "list" ],
"description": "The data published to the channel, sent as JSON to the
subscribed clients: text, a number or a boolean, a Map or a List. Bytes become
an array of numbers; a stream, a file or another object is sent as the text of
its toString: convert it first. The headers go in the ext field of the message"
},
+ "consumerOut": { "carriers": [ "text", "value", "map", "list", "null" ],
"description": "The data of the Bayeux message, parsed from JSON: a String, a
number or a boolean, a Map for an object, a List for an array, or null. From a
cometd producer in the same JVM the body that producer sent, as is (a stream or
a file as its text)" },
+ "consumerReply": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "Only when the exchange is InOut
(exchangePattern=InOut, or set in the route): the body the route ends with is
sent back to the client that published, as JSON. A stream or a file is read as
text, and null is sent as a JSON null. Bytes become an array of numbers;
another object is sent as the text of its toString" },
+ "producerIn": { "carriers": [ "text", "value", "map", "list", "stream",
"file", "null" ], "description": "The data published to the channel, sent as
JSON to the subscribed clients: text, a number or a boolean, a Map or a List. A
stream or a file is read as text, and null is published as a JSON null. Bytes
become an array of numbers; another object is sent as the text of its toString:
convert it first. The headers go in the ext field of the message" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdBinding.java
b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdBinding.java
index 814954d26253..28681ba00e78 100644
---
a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdBinding.java
+++
b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdBinding.java
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.cometd;
+import java.io.File;
+import java.io.InputStream;
+import java.io.Reader;
+import java.nio.file.Path;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
@@ -23,7 +27,10 @@ import java.util.Set;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
+import org.apache.camel.StreamCache;
+import org.apache.camel.WrappedFile;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.spi.Metadata;
import org.apache.camel.support.DefaultHeaderFilterStrategy;
@@ -35,6 +42,8 @@ import org.cometd.server.BayeuxServerImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.cometd.bayeux.Message.DATA_FIELD;
+
/**
* A Strategy used to convert between a Camel {@link Exchange} and to and from
a Cometd messages
*/
@@ -67,7 +76,8 @@ public class CometdBinding {
this.headerFilterStrategy = headerFilterStrategy;
}
- public ServerMessage.Mutable createCometdMessage(ServerChannel channel,
ServerSession serverSession, Message camelMessage) {
+ public ServerMessage.Mutable createCometdMessage(ServerChannel channel,
ServerSession serverSession, Message camelMessage)
+ throws InvalidPayloadException {
ServerMessage.Mutable mutable = bayeux.newMessage();
mutable.setChannel(channel.getId());
@@ -75,10 +85,26 @@ public class CometdBinding {
mutable.setClientId(serverSession.getId());
}
addHeadersToMessage(mutable, camelMessage);
- mutable.setData(camelMessage.getBody());
+ Object body = camelMessage.getBody();
+ if (body == null) {
+ // publish a JSON null: setData(null) leaves out the data field,
and the clients take a message without
+ // data for the reply of a publish, so their subscribers never
receive it
+ mutable.put(DATA_FIELD, null);
+ } else {
+ if (isStreamOrFile(body)) {
+ // the JSON of the message has the toString of a stream or a
file, so publish its content as text
+ body = camelMessage.getMandatoryBody(String.class);
+ }
+ mutable.setData(body);
+ }
return mutable;
}
+ private static boolean isStreamOrFile(Object body) {
+ return body instanceof StreamCache || body instanceof InputStream ||
body instanceof Reader
+ || body instanceof File || body instanceof Path || body
instanceof WrappedFile;
+ }
+
public Message createCamelMessage(
CamelContext camelContext, ServerSession remote, ServerMessage
cometdMessage, Object data) {
if (cometdMessage != null) {
diff --git
a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java
b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java
index e0a3c215dfbc..2c846c0b2d78 100644
---
a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java
+++
b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java
@@ -46,17 +46,24 @@ import org.apache.camel.util.ObjectHelper;
@BodyTypes(consumerOut = @BodyType(value = { Carrier.TEXT, Carrier.VALUE,
Carrier.MAP, Carrier.LIST, Carrier.NULL },
description = "The data of the Bayeux
message, parsed from JSON: a String, a number or a"
+ " boolean, a Map for an
object, a List for an array, or null. From a cometd"
- + " producer in the same JVM
the body that producer sent, as is"),
- consumerReply = @BodyType(value = { Carrier.TEXT, Carrier.VALUE,
Carrier.MAP, Carrier.LIST },
+ + " producer in the same JVM
the body that producer sent, as is (a stream or"
+ + " a file as its text)"),
+ consumerReply = @BodyType(value = {
+ Carrier.TEXT, Carrier.VALUE, Carrier.MAP, Carrier.LIST,
Carrier.STREAM,
+ Carrier.FILE, Carrier.NULL },
description = "Only when the exchange is
InOut (exchangePattern=InOut, or set in the"
+ " route): the body the
route ends with is sent back to the client that"
- + " published, as JSON.
Bytes become an array of numbers; a stream, a file"
- + " or another object is
sent as the text of its toString"),
- producerIn = @BodyType(value = { Carrier.TEXT, Carrier.VALUE,
Carrier.MAP, Carrier.LIST },
+ + " published, as JSON. A
stream or a file is read as text, and null is"
+ + " sent as a JSON null.
Bytes become an array of numbers; another object"
+ + " is sent as the text of
its toString"),
+ producerIn = @BodyType(value = {
+ Carrier.TEXT, Carrier.VALUE, Carrier.MAP, Carrier.LIST,
Carrier.STREAM,
+ Carrier.FILE, Carrier.NULL },
description = "The data published to the
channel, sent as JSON to the subscribed clients:"
- + " text, a number or a
boolean, a Map or a List. Bytes become an array of"
- + " numbers; a stream, a file
or another object is sent as the text of its"
- + " toString: convert it
first. The headers go in the ext field of the message"),
+ + " text, a number or a
boolean, a Map or a List. A stream or a file is read"
+ + " as text, and null is
published as a JSON null. Bytes become an array of"
+ + " numbers; another object is
sent as the text of its toString: convert it"
+ + " first. The headers go in
the ext field of the message"),
producerOut = @BodyType(Carrier.UNCHANGED))
public class CometdEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
diff --git
a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java
b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java
index e347975d24ee..9a0ff5a8caac 100644
---
a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java
+++
b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java
@@ -68,7 +68,7 @@ public class CometdProducer extends DefaultProducer
implements CometdProducerCon
}
@Override
- public void process(final Exchange exchange) {
+ public void process(final Exchange exchange) throws Exception {
service.process(exchange);
}
@@ -104,7 +104,7 @@ public class CometdProducer extends DefaultProducer
implements CometdProducerCon
this.disconnectLocalSession = disconnectLocalSession;
}
- public void process(final Exchange exchange) {
+ public void process(final Exchange exchange) throws Exception {
String channelName = producer.getEndpoint().getPath();
BayeuxServerImpl bayeux = producer.getBayeux();
ServerChannel channel = bayeux.getChannel(channelName);
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerBodyTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerBodyTest.java
new file mode 100644
index 000000000000..2140027a0aa6
--- /dev/null
+++
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerBodyTest.java
@@ -0,0 +1,201 @@
+/*
+ * 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.cometd;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.converter.stream.ByteArrayInputStreamCache;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.cometd.bayeux.Message;
+import org.cometd.client.BayeuxClient;
+import org.cometd.client.http.jetty.JettyHttpClientTransport;
+import org.eclipse.jetty.client.HttpClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The body a cometd producer publishes, and the body a cometd consumer
replies with, as a BayeuxClient receives them.
+ */
+public class CometdProducerBodyTest extends CamelTestSupport {
+
+ private static final String CHANNEL = "/test/body";
+ private static final String SERVICE_CHANNEL = "/service/reply";
+
+ @RegisterExtension
+ AvailablePortFinder.Port port = AvailablePortFinder.find();
+
+ @TempDir
+ Path tempDir;
+
+ private final BlockingQueue<Message> received = new
LinkedBlockingQueue<>();
+ private HttpClient httpClient;
+ private BayeuxClient client;
+
+ @BeforeEach
+ void connectClient() throws Exception {
+ httpClient = new HttpClient();
+ httpClient.start();
+ client = new BayeuxClient(
+ "http://127.0.0.1:"; + port.getPort() + "/cometd", new
JettyHttpClientTransport(null, httpClient));
+ client.handshake();
+ assertTrue(client.waitFor(5000, BayeuxClient.State.CONNECTED));
+
+ CountDownLatch subscribed = new CountDownLatch(2);
+ client.getChannel(CHANNEL).subscribe((channel, message) ->
received.add(message),
+ message -> subscribed.countDown());
+ client.getChannel(SERVICE_CHANNEL).subscribe((channel, message) ->
received.add(message),
+ message -> subscribed.countDown());
+ assertTrue(subscribed.await(5, TimeUnit.SECONDS), "Should subscribe
within timeout");
+ }
+
+ @AfterEach
+ void disconnectClient() throws Exception {
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ httpClient.stop();
+ }
+
+ @Test
+ void testTextBody() throws Exception {
+ template.sendBody("direct:input", "Hello World");
+
+ assertEquals("Hello World", receive().getData());
+ }
+
+ @Test
+ void testInputStreamBody() throws Exception {
+ template.sendBody("direct:input", new ByteArrayInputStream("Hello
Stream".getBytes(StandardCharsets.UTF_8)));
+
+ assertEquals("Hello Stream", receive().getData());
+ }
+
+ @Test
+ void testStreamCacheBody() throws Exception {
+ template.sendBody("direct:input", new ByteArrayInputStreamCache(
+ new ByteArrayInputStream("Hello Stream
Cache".getBytes(StandardCharsets.UTF_8))));
+
+ assertEquals("Hello Stream Cache", receive().getData());
+ }
+
+ @Test
+ void testReaderBody() throws Exception {
+ template.sendBody("direct:input", new StringReader("Hello Reader"));
+
+ assertEquals("Hello Reader", receive().getData());
+ }
+
+ @Test
+ void testFileBody() throws Exception {
+ Path file = Files.writeString(tempDir.resolve("hello.txt"), "Hello
File");
+
+ template.sendBody("direct:input", file.toFile());
+ assertEquals("Hello File", receive().getData());
+
+ template.sendBody("direct:input", file);
+ assertEquals("Hello File", receive().getData());
+ }
+
+ @Test
+ void testFileConsumerBody() throws Exception {
+ Files.writeString(tempDir.resolve("hello.txt"), "Hello File Consumer");
+
+ context.getRouteController().startRoute("file");
+
+ assertEquals("Hello File Consumer", receive().getData());
+ }
+
+ @Test
+ void testNullBody() throws Exception {
+ template.sendBody("direct:input", null);
+
+ Message message = receive();
+ assertEquals(CHANNEL, message.getChannel());
+ assertTrue(message.containsKey(Message.DATA_FIELD));
+ assertNull(message.getData());
+ }
+
+ @Test
+ void testNullReply() throws Exception {
+ client.getChannel(SERVICE_CHANNEL).publish("Hello");
+
+ Message message = receive();
+ assertEquals(SERVICE_CHANNEL, message.getChannel());
+ assertTrue(message.containsKey(Message.DATA_FIELD));
+ assertNull(message.getData());
+ }
+
+ @Test
+ void testStreamReply() throws Exception {
+ client.getChannel(SERVICE_CHANNEL).publish("stream");
+
+ assertEquals("Reply stream", receive().getData());
+ }
+
+ private Message receive() throws InterruptedException {
+ Message message = received.poll(5, TimeUnit.SECONDS);
+ assertNotNull(message, "Should receive a message within timeout");
+ return message;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ String options =
"?baseResource=file:./target/test-classes/webapp&timeout=240000&interval=0&maxInterval=30000"
+ +
"&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+ String producer = "cometd://127.0.0.1:" + port.getPort() + CHANNEL +
options;
+ String consumer = "cometd://127.0.0.1:" + port.getPort() +
SERVICE_CHANNEL + options;
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:input").to(producer);
+
+ File dir = tempDir.toFile();
+ fromF("file:%s?noop=true&initialDelay=0&delay=10",
dir.getAbsolutePath()).routeId("file").autoStartup(false)
+ .to(producer);
+
+ from(consumer)
+ .setExchangePattern(ExchangePattern.InOut)
+ .choice()
+ .when(body().isEqualTo("stream"))
+ .process(exchange -> exchange.getMessage().setBody(
+ new ByteArrayInputStream("Reply
stream".getBytes(StandardCharsets.UTF_8))))
+ .otherwise()
+ .process(exchange ->
exchange.getMessage().setBody(null));
+ }
+ };
+ }
+}
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 66d2601aecdf..0017d0fa6bfa 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
@@ -745,6 +745,16 @@ The `coap` consumer answered such a reply body with `2.05
Content` and an empty
Marshal such a body first (for example to JSON) or convert it to a `String`.
+=== camel-cometd - stream, file and null bodies are published
+
+The cometd producer, and the reply of an InOut cometd consumer, now publish a
stream, a `Reader` or a file body
+(`File`, `Path` or the body of the file consumer) as its content as text.
Before, the clients received the text of its
+`toString`, such as `ByteArrayInputStreamCache@...` or the path of the file. A
cometd consumer in the same JVM now
+receives such a body as a `String`.
+
+A `null` body is now published as a JSON `null` that the subscribed clients
receive. Before, the message had no data,
+so the clients took it for the reply of a publish and never received it, while
the exchange succeeded.
+
=== camel-aws - the HTTP client is no longer looked up on the classpath
The camel-aws components now set the HTTP client of the AWS SDK clients they
create for their endpoints: the Apache 5
