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 c4ba416b02f1 CAMEL-26137: camel-netty-http - reply to a List body and
to the request body sent back with disableStreamCache, keep the message with
sync=false (#29103)
c4ba416b02f1 is described below
commit c4ba416b02f11c5a8619dde6e9c69386dd9237fb
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 18:08:31 2026 +0100
CAMEL-26137: camel-netty-http - reply to a List body and to the request
body sent back with disableStreamCache, keep the message with sync=false
(#29103)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/components/netty-http.json | 2 +-
.../apache/camel/catalog/docs/netty-component.adoc | 2 +-
.../camel/component/netty/http/netty-http.json | 2 +-
.../netty/http/DefaultNettyHttpBinding.java | 23 ++-
.../component/netty/http/NettyHttpEndpoint.java | 4 +-
.../netty/http/NettyHttpProducerSyncFalseTest.java | 59 ++++++++
.../netty/http/NettyHttpReplyBodyTest.java | 164 +++++++++++++++++++++
.../camel-netty/src/main/docs/netty-component.adoc | 2 +-
.../camel/component/netty/NettyProducer.java | 22 +--
.../netty/NettyProducerSyncFalseReplyTest.java | 14 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 +-
11 files changed, 276 insertions(+), 23 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json
index f80cea968801..cfab0670749c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json
@@ -133,7 +133,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "stream" ], "javaType":
"org.apache.camel.component.netty.http.NettyChannelBufferStreamCache",
"dependsOn": [ "nettyHttpBinding" ], "byOption": [ { "option":
"disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] }, {
"option": "protocol", "value": "proxy", "carriers": [ "pojo" ], "javaType":
"io.netty.buffer.ByteBuf" } ], "description": "The request body as the stream
cache of Camel, which can be read many times; empty for a request wi [...]
- "consumerReply": { "carriers": [ "null", "text", "bytes", "stream",
"file", "value" ], "description": "The response body: text, bytes, a stream, a
file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map or a POJO is
written as its toString() and a List is not written at all: marshal it first. A
failed exchange sends 500 with an empty body (muteException)" },
+ "consumerReply": { "carriers": [ "null", "text", "bytes", "stream",
"file", "value" ], "description": "The response body: text, bytes, a stream, a
file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map, a List or a
POJO is written as its toString(): marshal it first. A failed exchange sends
500 with an empty body (muteException)" },
"producerIn": { "carriers": [ "null", "text", "bytes", "stream", "file" ],
"description": "The request body: text, bytes, a stream or a file, or a Netty
ByteBuf or HttpRequest. A Map, a List or a POJO fails: marshal it first.
Without the CamelHttpMethod header, no body sends a GET and a body a POST" },
"producerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"dependsOn": [ "nettyHttpBinding", "sync" ], "byOption": [ { "option":
"disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] } ],
"description": "The response body as a byte array, empty for a response without
a body. With disableStreamCache the Netty ByteBuf of the response, or an
InputStream for a chunked response; the ByteBuf is released when the exchange
is done" }
},
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc
index 8a8a558565ba..f270d1ad9dd8 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc
@@ -168,7 +168,7 @@ consumer closes the TCP channel, so the client does not
wait for a reply that ne
With `useByteBuf=true` a body that cannot be converted to a `ByteBuf` fails
the exchange instead of sending nothing.
With `sync=false` the producer continues routing as soon as the request is
written, and ignores any message the remote
-server sends back.
+server sends back. The message of the exchange stays as it was, also for an
`InOut` exchange.
==== Using Multiple Codecs
diff --git
a/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json
b/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json
index f80cea968801..cfab0670749c 100644
---
a/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json
+++
b/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json
@@ -133,7 +133,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "stream" ], "javaType":
"org.apache.camel.component.netty.http.NettyChannelBufferStreamCache",
"dependsOn": [ "nettyHttpBinding" ], "byOption": [ { "option":
"disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] }, {
"option": "protocol", "value": "proxy", "carriers": [ "pojo" ], "javaType":
"io.netty.buffer.ByteBuf" } ], "description": "The request body as the stream
cache of Camel, which can be read many times; empty for a request wi [...]
- "consumerReply": { "carriers": [ "null", "text", "bytes", "stream",
"file", "value" ], "description": "The response body: text, bytes, a stream, a
file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map or a POJO is
written as its toString() and a List is not written at all: marshal it first. A
failed exchange sends 500 with an empty body (muteException)" },
+ "consumerReply": { "carriers": [ "null", "text", "bytes", "stream",
"file", "value" ], "description": "The response body: text, bytes, a stream, a
file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map, a List or a
POJO is written as its toString(): marshal it first. A failed exchange sends
500 with an empty body (muteException)" },
"producerIn": { "carriers": [ "null", "text", "bytes", "stream", "file" ],
"description": "The request body: text, bytes, a stream or a file, or a Netty
ByteBuf or HttpRequest. A Map, a List or a POJO fails: marshal it first.
Without the CamelHttpMethod header, no body sends a GET and a body a POST" },
"producerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"dependsOn": [ "nettyHttpBinding", "sync" ], "byOption": [ { "option":
"disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] } ],
"description": "The response body as a byte array, empty for a response without
a body. With disableStreamCache the Netty ByteBuf of the response, or an
InputStream for a chunked response; the ByteBuf is released when the exchange
is done" }
},
diff --git
a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
index dd73344dc665..b6f1fde72d76 100644
---
a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
+++
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
@@ -502,6 +502,8 @@ public class DefaultNettyHttpBinding implements
NettyHttpBinding, Cloneable {
}
HttpResponse response = null;
+ // the request body (with disableStreamCache) when the reply sends it
back
+ ByteBuf requestBody = null;
if (body instanceof InputStream inputStream &&
configuration.isDisableStreamCache()) {
response = new OutboundStreamHttpResponse(
@@ -511,7 +513,14 @@ public class DefaultNettyHttpBinding implements
NettyHttpBinding, Cloneable {
if (response == null) {
if (body instanceof ByteBuf byteBuf) {
- buffer = byteBuf;
+ if (isRequestBody(message, byteBuf)) {
+ // the request body is released when the exchange is done,
and writing the response releases it
+ // too, so the reply writes a duplicate that gets a
reference of its own once the response is built
+ requestBody = byteBuf;
+ buffer = byteBuf.duplicate();
+ } else {
+ buffer = byteBuf;
+ }
} else {
// a String body is written in the charset of the content type
buffer = toByteBufWithContentTypeCharset(message, body);
@@ -602,9 +611,21 @@ public class DefaultNettyHttpBinding implements
NettyHttpBinding, Cloneable {
}
LOG.trace("Connection: {}", connection);
+ if (requestBody != null) {
+ // taken last, so a response that fails to build (such as a header
Netty rejects) keeps no reference
+ requestBody.retain();
+ }
return response;
}
+ /**
+ * Whether the buffer is the body of the HTTP request of the exchange.
+ */
+ private static boolean isRequestBody(Message message, ByteBuf buffer) {
+ return message.getExchange().getIn() instanceof NettyHttpMessage in &&
in.getHttpRequest() != null
+ && in.getHttpRequest().content() == buffer;
+ }
+
@Override
public HttpRequest toNettyRequest(Message message, String fullUri,
NettyHttpConfiguration configuration) throws Exception {
LOG.trace("toNettyRequest: {}", message);
diff --git
a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
index c160c004622c..474e004e1500 100644
---
a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
+++
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
@@ -71,8 +71,8 @@ import org.slf4j.LoggerFactory;
Carrier.NULL, Carrier.TEXT, Carrier.BYTES, Carrier.STREAM,
Carrier.FILE,
Carrier.VALUE },
description = "The response body: text,
bytes, a stream, a file, or a Netty ByteBuf or"
- + " HttpResponse; no body
sends 204. A Map or a POJO is written as its"
- + " toString() and a List
is not written at all: marshal it first. A"
+ + " HttpResponse; no body
sends 204. A Map, a List or a POJO is written as"
+ + " its toString(): marshal
it first. A"
+ " failed exchange sends
500 with an empty body (muteException)"),
producerIn = @BodyType(value = { Carrier.NULL, Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.FILE },
description = "The request body: text,
bytes, a stream or a file, or a Netty ByteBuf or"
diff --git
a/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpProducerSyncFalseTest.java
b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpProducerSyncFalseTest.java
new file mode 100644
index 000000000000..2ed77ea45338
--- /dev/null
+++
b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpProducerSyncFalseTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.netty.http;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * With sync=false the producer does not wait for the response, and the
response that arrives later is not set on the
+ * exchange.
+ */
+public class NettyHttpProducerSyncFalseTest extends BaseNettyTestSupport {
+
+ @Test
+ public void testResponseIgnored() throws Exception {
+ MockEndpoint server = getMockEndpoint("mock:server");
+ server.expectedBodiesReceived("Hello World");
+
+ Exchange exchange =
template.request("netty-http:http://localhost:{{port}}/foo?sync=false";,
+ e -> e.getIn().setBody("Hello World"));
+
+ MockEndpoint.assertIsSatisfied(context);
+ // the response the server sends back must not replace the message of
the exchange that already completed
+ await().during(500, TimeUnit.MILLISECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> "Hello
World".equals(exchange.getMessage().getBody(String.class)));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("netty-http:http://0.0.0.0:{{port}}/foo";)
+ .to("mock:server")
+ .setBody().constant("Bye World");
+ }
+ };
+ }
+}
diff --git
a/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpReplyBodyTest.java
b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpReplyBodyTest.java
new file mode 100644
index 000000000000..a2a05ed6da2d
--- /dev/null
+++
b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpReplyBodyTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.netty.http;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.TypeConversionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.support.TypeConverterSupport;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.classic.methods.HttpPost;
+import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.io.entity.StringEntity;
+import org.apache.hc.core5.util.Timeout;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The consumer writes a reply for a List body, and for the request body sent
back as-is with disableStreamCache=true. A
+ * reply that cannot be built is answered with a 500.
+ */
+public class NettyHttpReplyBodyTest extends BaseNettyTestSupport {
+
+ private final AtomicReference<ByteBuf> requestBody = new
AtomicReference<>();
+
+ @Test
+ public void testListReply() throws Exception {
+ Reply reply = execute(new HttpGet("http://localhost:"; + getPort() +
"/list"));
+ assertEquals(200, reply.code());
+ assertEquals("[a, b]", reply.body());
+ }
+
+ @Test
+ public void testFailingByteArrayConverterReply() throws Exception {
+ // a converter to byte[] that fails is not hidden behind the String
form of the body
+ Reply reply = execute(new HttpGet("http://localhost:"; + getPort() +
"/thing"));
+ assertEquals(500, reply.code());
+ }
+
+ @Test
+ public void testEchoRequestBodyWithDisableStreamCache() throws Exception {
+ for (int i = 0; i < 3; i++) {
+ Reply reply = execute(post("/echo", "Hello World " + i));
+ assertEquals(200, reply.code());
+ assertEquals("Hello World " + i, reply.body());
+ }
+ assertRequestBodyReleased();
+ }
+
+ @Test
+ public void testEchoRequestBodyFromAnotherThreadWithDisableStreamCache()
throws Exception {
+ for (int i = 0; i < 3; i++) {
+ Reply reply = execute(post("/echoThreads", "Hello World " + i));
+ assertEquals(200, reply.code());
+ assertEquals("Hello World " + i, reply.body());
+ }
+ assertRequestBodyReleased();
+ }
+
+ @Test
+ public void testEchoRequestBodyWithBadHeaderWithDisableStreamCache()
throws Exception {
+ // Netty rejects the header, so no response can be built, and the
request buffer must still be released
+ for (int i = 0; i < 3; i++) {
+ Reply reply = execute(post("/echoBadHeader", "Hello World " + i));
+ assertEquals(500, reply.code());
+ assertRequestBodyReleased();
+ }
+ }
+
+ private void assertRequestBodyReleased() {
+ // the reply writes its own reference of the request buffer, so the
buffer is still released once
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
requestBody.get().refCnt() == 0);
+ }
+
+ private HttpPost post(String path, String body) {
+ HttpPost post = new HttpPost("http://localhost:"; + getPort() + path);
+ post.setEntity(new StringEntity(body, ContentType.TEXT_PLAIN));
+ return post;
+ }
+
+ private Reply execute(HttpUriRequestBase request) throws Exception {
+ // before, the consumer sent no response at all, so do not wait for
one forever
+ RequestConfig config =
RequestConfig.custom().setResponseTimeout(Timeout.of(10,
TimeUnit.SECONDS)).build();
+ try (CloseableHttpClient client =
HttpClients.custom().setDefaultRequestConfig(config).build();
+ CloseableHttpResponse response = client.execute(request)) {
+ return new Reply(response.getCode(),
EntityUtils.toString(response.getEntity(), "UTF-8"));
+ }
+ }
+
+ private record Reply(int code, String body) {
+ }
+
+ private static final class Thing {
+ @Override
+ public String toString() {
+ return "Thing";
+ }
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getTypeConverterRegistry().addTypeConverter(byte[].class,
Thing.class, new TypeConverterSupport() {
+ @Override
+ public <T> T convertTo(Class<T> type, Exchange exchange, Object
value) {
+ throw new TypeConversionException(value, type, new
IOException("disk on fire"));
+ }
+ });
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("netty-http:http://0.0.0.0:{{port}}/list";)
+ .process(exchange ->
exchange.getMessage().setBody(List.of("a", "b")));
+
+ from("netty-http:http://0.0.0.0:{{port}}/thing";)
+ .process(exchange -> exchange.getMessage().setBody(new
Thing()));
+
+
from("netty-http:http://0.0.0.0:{{port}}/echoBadHeader?disableStreamCache=true";)
+ .process(exchange ->
requestBody.set(exchange.getMessage().getBody(ByteBuf.class)))
+ .setHeader("X-Bad", constant("line one\nline two"));
+
+
from("netty-http:http://0.0.0.0:{{port}}/echo?disableStreamCache=true";)
+ .process(exchange ->
requestBody.set(exchange.getMessage().getBody(ByteBuf.class)));
+
+
from("netty-http:http://0.0.0.0:{{port}}/echoThreads?disableStreamCache=true";)
+ .process(exchange ->
requestBody.set(exchange.getMessage().getBody(ByteBuf.class)))
+ .threads(1);
+ }
+ };
+ }
+}
diff --git a/components/camel-netty/src/main/docs/netty-component.adoc
b/components/camel-netty/src/main/docs/netty-component.adoc
index 8a8a558565ba..f270d1ad9dd8 100644
--- a/components/camel-netty/src/main/docs/netty-component.adoc
+++ b/components/camel-netty/src/main/docs/netty-component.adoc
@@ -168,7 +168,7 @@ consumer closes the TCP channel, so the client does not
wait for a reply that ne
With `useByteBuf=true` a body that cannot be converted to a `ByteBuf` fails
the exchange instead of sending nothing.
With `sync=false` the producer continues routing as soon as the request is
written, and ignores any message the remote
-server sends back.
+server sends back. The message of the exchange stays as it was, also for an
`InOut` exchange.
==== Using Multiple Codecs
diff --git
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
index af53bd2a532f..ca303308e963 100644
---
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
+++
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
@@ -59,7 +59,6 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.spi.CamelLogger;
import org.apache.camel.support.DefaultAsyncProducer;
-import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.IOHelper;
@@ -304,12 +303,8 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void onComplete(Exchange exchange) {
// should channel be closed after complete?
- Boolean close;
- if (ExchangeHelper.isOutCapable(exchange)) {
- close =
exchange.getOut().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE,
Boolean.class);
- } else {
- close =
exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE,
Boolean.class);
- }
+ Boolean close
+ =
exchange.getMessage().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE,
Boolean.class);
// should we disconnect, the header can override the
configuration
boolean disconnect = getConfiguration().isDisconnect();
@@ -390,14 +385,11 @@ public class NettyProducer extends DefaultAsyncProducer {
// if we do not expect any reply then signal callback to
continue routing
if (!configuration.isSync()) {
try {
- // should channel be closed after complete?
- Boolean close;
- if (ExchangeHelper.isOutCapable(exchange)) {
- close =
exchange.getOut().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE,
- Boolean.class);
- } else {
- close =
exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE,
Boolean.class);
- }
+ // should channel be closed after complete? (no reply
is received, so do not create an empty
+ // out message on an InOut exchange, which would
replace the message the exchange continues with)
+ Boolean close
+ =
exchange.getMessage().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE,
+ Boolean.class);
// should we disconnect, the header can override the
configuration
boolean disconnect = getConfiguration().isDisconnect();
diff --git
a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java
index b463a6cd0f2b..c4dade508389 100644
---
a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java
+++
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java
@@ -23,6 +23,7 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.Test;
import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* With sync=false the producer continues routing as soon as the request is
written, so a reply that the remote server
@@ -41,6 +42,19 @@ class NettyProducerSyncFalseReplyTest extends BaseNettyTest {
MockEndpoint.assertIsSatisfied(context);
}
+ @Test
+ void anInOutExchangeKeepsItsMessage() throws Exception {
+ getMockEndpoint("mock:server").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:result").setAssertPeriod(500);
+
+ // no reply is received, so the producer must not leave an empty out
message on an InOut exchange
+ Object reply = template.requestBody("direct:start", "Hello World");
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals("Hello World", reply);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
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 9deef6cc1d84..e386d7eacac7 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
@@ -2241,7 +2241,8 @@ also converts a reply set as the out message to a
`ByteBuf`. `NettyPayloadHelper
`InvalidPayloadException`.
With `sync=false` the producer now ignores a message the remote server sends
back. Before, that reply was set on the
-exchange after it had already continued routing.
+exchange after it had already continued routing. The producer now also leaves
the body and headers of an `InOut`
+exchange as they were. Before, the exchange continued with an empty message.
This also applies to `netty-http`.
=== camel-netty - a reply that cannot be written, and allowSerializedHeaders
on the reply
@@ -2275,6 +2276,8 @@ When no HTTP response can be built from the reply of the
route (for example the
now answers with HTTP Status 500 and closes the connection. The body is empty,
or the stack trace of the exception with
`muteException=false`. Before, no response was sent and the client waited
until it timed out.
+A `List` reply is now written as its `toString()`, as a `Map` or a POJO
already was. Before, no response was sent.
+
=== camel-odata - the body of CREATE and UPDATE, and a response that is not a
JSON object
`CREATE` and `UPDATE` send JSON text given as a byte array, a stream or a file
as it is, as they do for a `String`.
