This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/29165-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 76507b4ab7ce2d22858f5c2ee5cb842fb36c8d5b Author: Guillaume Nodet <[email protected]> AuthorDate: Wed Nov 4 19:04:14 2026 +0000 [backport camel-4.18.x] CAMEL-26587: camel-netty - check transferExchange before processing transferred exchange objects --- .../camel/component/netty/NettyPayloadHelper.java | 67 ++++++++- .../netty/handlers/ClientChannelHandler.java | 4 +- .../netty/handlers/ServerChannelHandler.java | 2 +- .../netty/NettyTransferExchangeDisabledTest.java | 149 +++++++++++++++++++++ 4 files changed, 213 insertions(+), 9 deletions(-) diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java index 9fd495930dee..5c13f39e860d 100644 --- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java +++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java @@ -63,14 +63,23 @@ public final class NettyPayloadHelper { } } - public static void setIn(Exchange exchange, Object payload) { - if (payload instanceof DefaultExchangeHolder) { + /** + * Sets the inbound message of the exchange from a decoded Netty payload. A {@link DefaultExchangeHolder} rebuilds + * the exchange only when the endpoint has {@code transferExchange=true}; otherwise it is set as the message body + * like any other payload. An {@link AddressedEnvelope} (UDP) is unwrapped and its sender address is recorded. + * + * @param endpoint the endpoint that received the payload, whose configuration decides on the exchange transfer + * @param exchange the exchange to set the message on + * @param payload the decoded payload + */ + public static void setIn(NettyEndpoint endpoint, Exchange exchange, Object payload) { + if (isExchangeTransfer(endpoint, payload)) { unmarshal(exchange, (DefaultExchangeHolder) payload); } else if (payload instanceof AddressedEnvelope) { @SuppressWarnings("unchecked") AddressedEnvelope<Object, InetSocketAddress> dp = (AddressedEnvelope<Object, InetSocketAddress>) payload; // need to check if the content is ExchangeHolder - if (dp.content() instanceof DefaultExchangeHolder) { + if (isExchangeTransfer(endpoint, dp.content())) { unmarshal(exchange, (DefaultExchangeHolder) dp.content()); } else { // need to take out the payload here @@ -86,14 +95,22 @@ public final class NettyPayloadHelper { } } - public static void setOut(Exchange exchange, Object payload) { - if (payload instanceof DefaultExchangeHolder) { + /** + * Sets the outbound message of the exchange from a decoded Netty payload, as {@link #setIn} does for the inbound + * message; a payload that is a plain body keeps the headers of the inbound message. + * + * @param endpoint the endpoint that received the payload, whose configuration decides on the exchange transfer + * @param exchange the exchange to set the message on + * @param payload the decoded payload + */ + public static void setOut(NettyEndpoint endpoint, Exchange exchange, Object payload) { + if (isExchangeTransfer(endpoint, payload)) { unmarshal(exchange, (DefaultExchangeHolder) payload); } else if (payload instanceof AddressedEnvelope) { @SuppressWarnings("unchecked") AddressedEnvelope<Object, InetSocketAddress> dp = (AddressedEnvelope<Object, InetSocketAddress>) payload; // need to check if the content is ExchangeHolder - if (dp.content() instanceof DefaultExchangeHolder) { + if (isExchangeTransfer(endpoint, dp.content())) { unmarshal(exchange, (DefaultExchangeHolder) dp.content()); } else { // need to take out the payload here @@ -108,6 +125,44 @@ public final class NettyPayloadHelper { } } + /** + * A decoded {@link DefaultExchangeHolder} rebuilds the entire Exchange (id, body, headers, out message, properties + * and variables), so it is only unmarshalled when the endpoint has explicitly enabled exchange transfer via + * {@code transferExchange=true}. Otherwise the payload is treated as a normal message body. + */ + private static boolean isExchangeTransfer(NettyEndpoint endpoint, Object payload) { + return payload instanceof DefaultExchangeHolder + && endpoint != null && endpoint.getConfiguration().isTransferExchange(); + } + + /** + * @deprecated use {@link #setIn(NettyEndpoint, Exchange, Object)}. This overload has no endpoint and so + * cannot check {@code transferExchange}: it never unmarshals a {@link DefaultExchangeHolder} + * and sets any payload, a holder included, as the message body. A caller that relied on the + * unconditional unmarshalling of earlier versions has to pass its endpoint to the + * three-argument overload. + * @param exchange the exchange + * @param payload the decoded payload + */ + @Deprecated + public static void setIn(Exchange exchange, Object payload) { + setIn(null, exchange, payload); + } + + /** + * @deprecated use {@link #setOut(NettyEndpoint, Exchange, Object)}. This overload has no endpoint and so + * cannot check {@code transferExchange}: it never unmarshals a {@link DefaultExchangeHolder} + * and sets any payload, a holder included, as the message body. A caller that relied on the + * unconditional unmarshalling of earlier versions has to pass its endpoint to the + * three-argument overload. + * @param exchange the exchange + * @param payload the decoded payload + */ + @Deprecated + public static void setOut(Exchange exchange, Object payload) { + setOut(null, exchange, payload); + } + /** * The headers and exchange properties that describe the Netty connection of this side. */ diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java index 2205e096f57e..d98996cb1551 100644 --- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java +++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java @@ -257,10 +257,10 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<Object> { // set the result on either IN or OUT on the original exchange depending on its pattern if (ExchangeHelper.isOutCapable(exchange)) { - NettyPayloadHelper.setOut(exchange, body); + NettyPayloadHelper.setOut(producer.getEndpoint(), exchange, body); return exchange.getOut(); } else { - NettyPayloadHelper.setIn(exchange, body); + NettyPayloadHelper.setIn(producer.getEndpoint(), exchange, body); return exchange.getIn(); } } diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java index a056028b8345..05f108237933 100644 --- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java +++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java @@ -116,7 +116,7 @@ public class ServerChannelHandler extends SimpleChannelInboundHandler<Object> { // must be prototype scoped (not pooled) so we create the exchange via endpoint Exchange exchange = consumer.createExchange(false); consumer.getEndpoint().updateMessageHeader(exchange.getIn(), ctx); - NettyPayloadHelper.setIn(exchange, message); + NettyPayloadHelper.setIn(consumer.getEndpoint(), exchange, message); return exchange; } diff --git a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyTransferExchangeDisabledTest.java b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyTransferExchangeDisabledTest.java new file mode 100644 index 000000000000..829fff44be15 --- /dev/null +++ b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyTransferExchangeDisabledTest.java @@ -0,0 +1,149 @@ +/* + * 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; + +import io.netty.channel.ChannelHandler; +import io.netty.handler.codec.serialization.ClassResolvers; +import org.apache.camel.BindToRegistry; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.Producer; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.netty.codec.ObjectDecoder; +import org.apache.camel.component.netty.codec.ObjectEncoder; +import org.apache.camel.support.DefaultExchange; +import org.apache.camel.support.DefaultExchangeHolder; +import org.junit.jupiter.api.Test; + +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.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class NettyTransferExchangeDisabledTest extends BaseNettyTest { + + @BindToRegistry("encoder") + public ChannelHandler getEncoder() { + return new ShareableChannelHandlerFactory(new ObjectEncoder()); + } + + @BindToRegistry("decoder") + public ChannelHandler getDecoder() { + return new DefaultChannelHandlerFactory() { + @Override + public ChannelHandler newChannelHandler() { + return new ObjectDecoder(ClassResolvers.weakCachingResolver(null)); + } + }; + } + + @Test + public void testTransferExchangeDisabledTreatsHolderAsBody() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(1); + + Exchange sourceExchange = new DefaultExchange(context); + sourceExchange.getIn().setBody("Inner Payload"); + sourceExchange.getIn().setHeader("CustomHeader", "HeaderValue"); + sourceExchange.setProperty("CustomProperty", "PropertyValue"); + + DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(sourceExchange, true, true); + + Endpoint endpoint + = context.getEndpoint("netty:tcp://localhost:{{port}}?sync=false&encoders=#encoder&decoders=#decoder"); + Producer producer = endpoint.createProducer(); + producer.start(); + try { + Exchange sendExchange = endpoint.createExchange(); + sendExchange.getIn().setBody(holder); + producer.process(sendExchange); + } finally { + producer.stop(); + } + + mock.assertIsSatisfied(); + + Exchange received = mock.getReceivedExchanges().get(0); + assertNotNull(received); + + // Holder must remain the body and NOT be unmarshalled into the exchange + assertTrue(received.getIn().getBody() instanceof DefaultExchangeHolder, + "Body should remain a DefaultExchangeHolder when transferExchange=false"); + assertNull(received.getIn().getHeader("CustomHeader"), + "Headers from DefaultExchangeHolder must not be unmarshalled when transferExchange=false"); + assertNull(received.getProperty("CustomProperty"), + "Properties from DefaultExchangeHolder must not be unmarshalled when transferExchange=false"); + assertNotNull(received.getIn().getHeader(NettyConstants.NETTY_REMOTE_ADDRESS), + "Connection headers must be preserved"); + } + + @Test + public void testHelperGating() { + NettyEndpoint transferOn = context.getEndpoint("netty:tcp://localhost:1111?transferExchange=true", NettyEndpoint.class); + NettyEndpoint transferOff + = context.getEndpoint("netty:tcp://localhost:2222?transferExchange=false", NettyEndpoint.class); + + Exchange source = new DefaultExchange(context); + source.getIn().setBody("Inner Content"); + source.getIn().setHeader("Key", "Value"); + source.setProperty("Prop", "Val"); + DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(source, true, true); + + // 1. With transferExchange = true + Exchange targetOn = new DefaultExchange(context); + NettyPayloadHelper.setIn(transferOn, targetOn, holder); + assertEquals("Inner Content", targetOn.getIn().getBody()); + assertEquals("Value", targetOn.getIn().getHeader("Key")); + assertEquals("Val", targetOn.getProperty("Prop")); + + // 2. With transferExchange = false + Exchange targetOff = new DefaultExchange(context); + NettyPayloadHelper.setIn(transferOff, targetOff, holder); + assertSame(holder, targetOff.getIn().getBody()); + assertNull(targetOff.getIn().getHeader("Key")); + assertNull(targetOff.getProperty("Prop")); + + // 3. With null endpoint (deprecated overload) + Exchange targetNull = new DefaultExchange(context); + NettyPayloadHelper.setIn(targetNull, holder); + assertSame(holder, targetNull.getIn().getBody()); + assertNull(targetNull.getIn().getHeader("Key")); + assertNull(targetNull.getProperty("Prop")); + + // 4. setOut with transferExchange = false + Exchange targetOutOff = new DefaultExchange(context); + targetOutOff.getIn().setHeader("ExistingHeader", "ExistingVal"); + NettyPayloadHelper.setOut(transferOff, targetOutOff, holder); + assertSame(holder, targetOutOff.getMessage().getBody()); + assertNull(targetOutOff.getMessage().getHeader("Key")); + assertEquals("ExistingVal", targetOutOff.getMessage().getHeader("ExistingHeader")); + assertNull(targetOutOff.getProperty("Prop")); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("netty:tcp://localhost:{{port}}?sync=false&encoders=#encoder&decoders=#decoder") + .to("mock:result"); + } + }; + } +}
