This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26182-mina-sync in repository https://gitbox.apache.org/repos/asf/camel.git
commit 60f81cf44e85c7779aae9271d4c9a6e3ef44f9ec Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:49:16 2026 +0100 CAMEL-26182: camel-mina - sync=false is one-way: the producer keeps the message and the consumer writes no reply, and a textline consumer converts the reply to a String With sync=false the producer read the close-session header with exchange.getOut(), which replaced the message of an InOut exchange with an empty one. It now uses getMessage(). The consumer wrote the body back to the client also with sync=false; it now replies only with sync=true. With textline=true the consumer now converts the reply to a String, as the producer does, instead of writing a byte[] or stream as its toString(). The consumer now honours disconnectOnNoReply and noReplyLogLevel when there is no reply. The docs explain that a byte[] reply with the default object codec needs objectCodecPattern=[B on the producer, and the upgrade guide lists the changes. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/components/mina.json | 4 +- .../apache/camel/catalog/docs/mina-component.adoc | 18 ++- .../org/apache/camel/component/mina/mina.json | 4 +- .../camel-mina/src/main/docs/mina-component.adoc | 18 ++- .../apache/camel/component/mina/MinaConsumer.java | 61 +++++---- .../apache/camel/component/mina/MinaEndpoint.java | 22 ++-- .../apache/camel/component/mina/MinaProducer.java | 10 +- .../component/mina/MinaConsumerReplyTest.java | 146 +++++++++++++++++++++ .../camel/component/mina/MinaSyncFalseTest.java | 113 ++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 15 +++ 10 files changed, 362 insertions(+), 49 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mina.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mina.json index 10b6c4d37ed4..8ae54d3daea6 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mina.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mina.json @@ -63,9 +63,9 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String", "dependsOn": [ "codec", "objectCodecPattern", "transferExchange" ], "byOption": [ { "option": "protocol", "value": "udp", "carriers": [ "bytes" ], "javaType": "byte[]" }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "The data received as a String: one line with textline=true. The default object codec over TCP acc [...] - "consumerReply": { "carriers": [ "null", "text" ], "dependsOn": [ "codec", "transferExchange" ], "byOption": [ { "option": "protocol", "value": "udp", "carriers": [ "null", "text", "bytes", "stream", "file" ] }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "null", "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "The body the route ends with is written back to the caller, also with sync=false. Send a String: with textline=true other [...] + "consumerReply": { "carriers": [ "null", "text" ], "dependsOn": [ "codec", "transferExchange" ], "byOption": [ { "option": "textline", "value": "true", "carriers": [ "any" ] }, { "option": "protocol", "value": "udp", "carriers": [ "null", "text", "bytes", "stream", "file" ] }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "null", "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "With sync=true (the default) the body the route ends wi [...] "producerIn": { "carriers": [ "null", "text" ], "dependsOn": [ "codec", "transferExchange" ], "byOption": [ { "option": "textline", "value": "true", "carriers": [ "any" ] }, { "option": "protocol", "value": "udp", "carriers": [ "null", "text", "bytes", "stream", "file" ] }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "null", "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "The data to send: a String. The default object codec over [...] - "producerOut": { "carriers": [ "text", "null" ], "javaType": "java.lang.String", "dependsOn": [ "codec", "objectCodecPattern", "transferExchange" ], "byOption": [ { "option": "sync", "value": "false", "carriers": [ "unchanged", "null" ] }, { "option": "protocol", "value": "udp", "carriers": [ "bytes", "null" ], "javaType": "byte[]" }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "pojo", "null" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description [...] + "producerOut": { "carriers": [ "text", "null" ], "javaType": "java.lang.String", "dependsOn": [ "codec", "objectCodecPattern", "transferExchange" ], "byOption": [ { "option": "sync", "value": "false", "carriers": [ "unchanged" ] }, { "option": "protocol", "value": "udp", "carriers": [ "bytes", "null" ], "javaType": "byte[]" }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "pojo", "null" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "With [...] }, "properties": { "protocol": { "index": 0, "kind": "path", "displayName": "Protocol", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.mina.MinaConfiguration", "configurationField": "configuration", "description": "Protocol to use" }, diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mina-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mina-component.adoc index 0b4c3786da8b..ffe5f7d90073 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mina-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mina-component.adoc @@ -21,8 +21,9 @@ Favor using xref:netty-component.adoc[Netty] as Netty is a much more active maintained and popular project than Apache Mina currently is. ==== -WARNING: Be careful with `sync=false` on consumer endpoints. In camel-mina, all -consumer exchanges are `InOut`. This is different to camel-mina. +NOTE: With `sync=false` the endpoint is one-way: the consumer does not write a reply back to the client, and the +producer does not wait for a reply and leaves the message of the exchange as it was. Use `sync=true` (the default) for +request-reply. Maven users will need to add the following dependency to their `pom.xml` for this component: @@ -82,6 +83,19 @@ Registry; for example, by creating a bean in the Spring XML file. Then use the `codec` option to specify the bean ID of your codec. See xref:dataformats:hl7-dataformat.adoc[HL7] that has a custom codec. +=== The reply of the consumer + +With `sync=true` (the default) the consumer writes the body the route ends with back to the client. With +`textline=true` the body is converted to a `String` first, so a `byte[]` or a stream is written as its text. + +The default object codec (no `textline` and no `codec`) only deserializes the classes that `objectCodecPattern` lists, +and a `String`. This applies to the reply as well: a producer that should receive a `byte[]` reply needs +`objectCodecPattern=[B`. Without it the producer fails to decode the reply with a `StreamCorruptedException` (such as +`invalid type code: 42`). Keep `objectCodecPattern` as narrow as the replies need. + +When the route ends with no body, the consumer writes nothing, logs it at `noReplyLogLevel` and closes the session, so +the client does not wait for a reply. Set `disconnectOnNoReply=false` to keep the session open. + === Get the IoSession for message You can get the IoSession from the message header with this key diff --git a/components/camel-mina/src/generated/resources/META-INF/org/apache/camel/component/mina/mina.json b/components/camel-mina/src/generated/resources/META-INF/org/apache/camel/component/mina/mina.json index 10b6c4d37ed4..8ae54d3daea6 100644 --- a/components/camel-mina/src/generated/resources/META-INF/org/apache/camel/component/mina/mina.json +++ b/components/camel-mina/src/generated/resources/META-INF/org/apache/camel/component/mina/mina.json @@ -63,9 +63,9 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String", "dependsOn": [ "codec", "objectCodecPattern", "transferExchange" ], "byOption": [ { "option": "protocol", "value": "udp", "carriers": [ "bytes" ], "javaType": "byte[]" }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "The data received as a String: one line with textline=true. The default object codec over TCP acc [...] - "consumerReply": { "carriers": [ "null", "text" ], "dependsOn": [ "codec", "transferExchange" ], "byOption": [ { "option": "protocol", "value": "udp", "carriers": [ "null", "text", "bytes", "stream", "file" ] }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "null", "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "The body the route ends with is written back to the caller, also with sync=false. Send a String: with textline=true other [...] + "consumerReply": { "carriers": [ "null", "text" ], "dependsOn": [ "codec", "transferExchange" ], "byOption": [ { "option": "textline", "value": "true", "carriers": [ "any" ] }, { "option": "protocol", "value": "udp", "carriers": [ "null", "text", "bytes", "stream", "file" ] }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "null", "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "With sync=true (the default) the body the route ends wi [...] "producerIn": { "carriers": [ "null", "text" ], "dependsOn": [ "codec", "transferExchange" ], "byOption": [ { "option": "textline", "value": "true", "carriers": [ "any" ] }, { "option": "protocol", "value": "udp", "carriers": [ "null", "text", "bytes", "stream", "file" ] }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "null", "pojo" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "The data to send: a String. The default object codec over [...] - "producerOut": { "carriers": [ "text", "null" ], "javaType": "java.lang.String", "dependsOn": [ "codec", "objectCodecPattern", "transferExchange" ], "byOption": [ { "option": "sync", "value": "false", "carriers": [ "unchanged", "null" ] }, { "option": "protocol", "value": "udp", "carriers": [ "bytes", "null" ], "javaType": "byte[]" }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "pojo", "null" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description [...] + "producerOut": { "carriers": [ "text", "null" ], "javaType": "java.lang.String", "dependsOn": [ "codec", "objectCodecPattern", "transferExchange" ], "byOption": [ { "option": "sync", "value": "false", "carriers": [ "unchanged" ] }, { "option": "protocol", "value": "udp", "carriers": [ "bytes", "null" ], "javaType": "byte[]" }, { "option": "allowDefaultCodec", "value": "false", "carriers": [ "pojo", "null" ], "javaType": "org.apache.mina.core.buffer.IoBuffer" } ], "description": "With [...] }, "properties": { "protocol": { "index": 0, "kind": "path", "displayName": "Protocol", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.mina.MinaConfiguration", "configurationField": "configuration", "description": "Protocol to use" }, diff --git a/components/camel-mina/src/main/docs/mina-component.adoc b/components/camel-mina/src/main/docs/mina-component.adoc index 0b4c3786da8b..ffe5f7d90073 100644 --- a/components/camel-mina/src/main/docs/mina-component.adoc +++ b/components/camel-mina/src/main/docs/mina-component.adoc @@ -21,8 +21,9 @@ Favor using xref:netty-component.adoc[Netty] as Netty is a much more active maintained and popular project than Apache Mina currently is. ==== -WARNING: Be careful with `sync=false` on consumer endpoints. In camel-mina, all -consumer exchanges are `InOut`. This is different to camel-mina. +NOTE: With `sync=false` the endpoint is one-way: the consumer does not write a reply back to the client, and the +producer does not wait for a reply and leaves the message of the exchange as it was. Use `sync=true` (the default) for +request-reply. Maven users will need to add the following dependency to their `pom.xml` for this component: @@ -82,6 +83,19 @@ Registry; for example, by creating a bean in the Spring XML file. Then use the `codec` option to specify the bean ID of your codec. See xref:dataformats:hl7-dataformat.adoc[HL7] that has a custom codec. +=== The reply of the consumer + +With `sync=true` (the default) the consumer writes the body the route ends with back to the client. With +`textline=true` the body is converted to a `String` first, so a `byte[]` or a stream is written as its text. + +The default object codec (no `textline` and no `codec`) only deserializes the classes that `objectCodecPattern` lists, +and a `String`. This applies to the reply as well: a producer that should receive a `byte[]` reply needs +`objectCodecPattern=[B`. Without it the producer fails to decode the reply with a `StreamCorruptedException` (such as +`invalid type code: 42`). Keep `objectCodecPattern` as narrow as the replies need. + +When the route ends with no body, the consumer writes nothing, logs it at `noReplyLogLevel` and closes the session, so +the client does not wait for a reply. Set `disconnectOnNoReply=false` to keep the session open. + === Get the IoSession for message You can get the IoSession from the message header with this key diff --git a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java index cb9350edaea4..4550b8ae8a4f 100644 --- a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java +++ b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java @@ -29,6 +29,7 @@ import org.apache.camel.Exchange; import org.apache.camel.ExchangePattern; import org.apache.camel.ExchangePropertyKey; import org.apache.camel.Processor; +import org.apache.camel.spi.CamelLogger; import org.apache.camel.support.DefaultConsumer; import org.apache.camel.support.ExchangeHelper; import org.apache.camel.util.IOHelper; @@ -70,6 +71,7 @@ public class MinaConsumer extends DefaultConsumer { private SocketAddress address; private IoAcceptor acceptor; private final MinaConfiguration configuration; + private final CamelLogger noReplyLogger; private ExecutorService workerPool; // the acceptor or connector and the worker pool were released by doStop, and are created again on start private boolean released; @@ -77,6 +79,7 @@ public class MinaConsumer extends DefaultConsumer { public MinaConsumer(final MinaEndpoint endpoint, Processor processor) throws Exception { super(endpoint, processor); this.configuration = endpoint.getConfiguration(); + this.noReplyLogger = new CamelLogger(LOG, configuration.getNoReplyLogLevel()); // // All mina endpoints are InOut. The endpoints are asynchronous. // Endpoints can send "n" messages and receive "m" messages. @@ -491,34 +494,44 @@ public class MinaConsumer extends DefaultConsumer { } try { - // - // If there's a response to send, send it. - // boolean disconnect = getEndpoint().getConfiguration().isDisconnect(); - Object response; - if (exchange.hasOut()) { - response = MinaPayloadHelper.getOut(getEndpoint(), exchange); - } else { - response = MinaPayloadHelper.getIn(getEndpoint(), exchange); - } - - boolean failed = exchange.isFailed(); - if (failed && !getEndpoint().getConfiguration().isTransferExchange()) { - if (exchange.getException() != null) { - response = getEndpoint().getConfiguration().isMuteException() - ? mutedException() : exchange.getException(); + // only reply when sync is enabled (request-reply); with sync=false the endpoint is one-way + if (configuration.isSync()) { + Object response; + if (exchange.hasOut()) { + response = MinaPayloadHelper.getOut(getEndpoint(), exchange); } else { - // failed and no exception, must be a fault - response = exchange.getOut().getBody(); + response = MinaPayloadHelper.getIn(getEndpoint(), exchange); } - } - if (response != null) { - LOG.debug("Writing body: {}", response); - MinaHelper.writeBody(session, response, exchange, configuration.getWriteTimeout()); - } else { - LOG.debug("Writing no response"); - disconnect = Boolean.TRUE; + boolean failed = exchange.isFailed(); + if (failed && !getEndpoint().getConfiguration().isTransferExchange()) { + if (exchange.getException() != null) { + response = getEndpoint().getConfiguration().isMuteException() + ? mutedException() : exchange.getException(); + } else { + // failed and no exception, must be a fault + response = exchange.getOut().getBody(); + } + } + + if (response != null) { + // if textline enabled then convert to a String which must be used for textline + // (the text line encoder would otherwise write a byte[] or a stream with its toString) + if (configuration.isTextline()) { + response = getEndpoint().getCamelContext().getTypeConverter().mandatoryConvertTo(String.class, + exchange, response); + } + LOG.debug("Writing body: {}", response); + MinaHelper.writeBody(session, response, exchange, configuration.getWriteTimeout()); + } else { + noReplyLogger.log("No payload to send as reply for exchange: " + exchange); + if (configuration.isDisconnectOnNoReply()) { + // must close session if no data to write otherwise client will never receive a response + // and wait forever (if not timing out) + disconnect = true; + } + } } // should session be closed after complete? diff --git a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java index 0ce86273caf5..2b7b97f9f367 100644 --- a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java +++ b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java @@ -50,6 +50,7 @@ import org.apache.camel.util.ObjectHelper; consumerReply = @BodyType(value = { Carrier.NULL, Carrier.TEXT }, dependsOn = { "codec", "transferExchange" }, byOption = { + @BodyType.When(option = "textline", value = "true", carriers = Carrier.ANY), @BodyType.When(option = "protocol", value = "udp", carriers = { Carrier.NULL, Carrier.TEXT, Carrier.BYTES, @@ -57,12 +58,13 @@ import org.apache.camel.util.ObjectHelper; @BodyType.When(option = "allowDefaultCodec", value = "false", carriers = { Carrier.NULL, Carrier.POJO }, javaType = "org.apache.mina.core.buffer.IoBuffer") }, - description = "The body the route ends with is written back to the caller, also with" - + " sync=false. Send a String: with textline=true other bodies are" - + " written with toString(), and the object codec writes any" - + " Serializable object, which a Mina client accepts only when its" - + " objectCodecPattern lists the class. No body sends nothing and closes" - + " the connection"), + description = "With sync=true (the default) the body the route ends with is written" + + " back to the caller; with sync=false nothing is written. Send a" + + " String: textline=true converts any body to a line of text, and the" + + " object codec writes any Serializable object, which a Mina client" + + " accepts only when its objectCodecPattern lists the class (a byte" + + " array needs [B). No body sends nothing and closes the connection" + + " (disconnectOnNoReply)"), producerIn = @BodyType(value = { Carrier.NULL, Carrier.TEXT }, dependsOn = { "codec", "transferExchange" }, byOption = { @@ -81,16 +83,16 @@ import org.apache.camel.util.ObjectHelper; producerOut = @BodyType(value = { Carrier.TEXT, Carrier.NULL }, javaType = "java.lang.String", dependsOn = { "codec", "objectCodecPattern", "transferExchange" }, byOption = { - @BodyType.When(option = "sync", value = "false", - carriers = { Carrier.UNCHANGED, Carrier.NULL }), + @BodyType.When(option = "sync", value = "false", carriers = Carrier.UNCHANGED), @BodyType.When(option = "protocol", value = "udp", carriers = { Carrier.BYTES, Carrier.NULL }, javaType = "byte[]"), @BodyType.When(option = "allowDefaultCodec", value = "false", carriers = { Carrier.POJO, Carrier.NULL }, javaType = "org.apache.mina.core.buffer.IoBuffer") }, description = "With sync=true (the default) the reply of the server as a String, or a" - + " byte array over UDP. No body is not sent and stays null. With" - + " sync=false an InOut exchange ends with an empty message")) + + " byte array over UDP. The object codec accepts another reply only when" + + " objectCodecPattern lists its class (a byte array needs [B). No body is" + + " not sent and stays null. With sync=false the message is unchanged")) public class MinaEndpoint extends DefaultEndpoint implements MultipleConsumersSupport, EndpointServiceLocation { @UriParam diff --git a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java index 934b4e20db33..764ebd035ce4 100644 --- a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java +++ b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java @@ -211,13 +211,9 @@ public class MinaProducer extends DefaultProducer { return; } - // should session be closed after complete? - Boolean close; - if (ExchangeHelper.isOutCapable(exchange)) { - close = exchange.getOut().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); - } else { - close = exchange.getIn().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); - } + // should session be closed after complete? (use getMessage as there is no out message with sync=false, + // and calling getOut would replace the message with an empty one) + Boolean close = exchange.getMessage().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); // should we disconnect, the header can override the configuration boolean disconnect = getEndpoint().getConfiguration().isDisconnect(); diff --git a/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaConsumerReplyTest.java b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaConsumerReplyTest.java new file mode 100644 index 000000000000..9e1a6a0f21c5 --- /dev/null +++ b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaConsumerReplyTest.java @@ -0,0 +1,146 @@ +/* + * 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.mina; + +import java.io.BufferedReader; +import java.io.ByteArrayInputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; + +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * The reply a consumer with sync=true writes back: a textline reply that is not a String, the + * CamelMinaCloseSessionWhenComplete header, disconnectOnNoReply, and a byte[] reply with the object codec. + */ +public class MinaConsumerReplyTest extends BaseMinaTest { + + private int textlinePort; + private int noReplyPort; + private int objectPort; + + @Test + public void testTextlineBytesReply() { + Object out = template.requestBody(String.format("mina:tcp://localhost:%d?textline=true", textlinePort), "bytes"); + assertEquals("Bye bytes", out); + } + + @Test + public void testTextlineStreamReply() { + Object out = template.requestBody(String.format("mina:tcp://localhost:%d?textline=true", textlinePort), "stream"); + assertEquals("Bye stream", out); + } + + @Test + public void testCloseSessionWhenCompleteHeader() throws Exception { + try (Socket socket = connect(textlinePort)) { + BufferedReader reader = send(socket, "close"); + assertEquals("Bye close", reader.readLine()); + // the header set by the route closes the session + assertNull(reader.readLine()); + } + } + + @Test + public void testSessionStaysOpenWithoutCloseHeader() throws Exception { + try (Socket socket = connect(textlinePort)) { + BufferedReader reader = send(socket, "bytes"); + assertEquals("Bye bytes", reader.readLine()); + assertThrows(SocketTimeoutException.class, reader::readLine); + } + } + + @Test + public void testDisconnectOnNoReplyFalse() throws Exception { + try (Socket socket = connect(noReplyPort)) { + // no reply, and the session stays open + BufferedReader reader = send(socket, "London"); + assertThrows(SocketTimeoutException.class, reader::readLine); + + // so the next request on the same session gets its reply + reader = send(socket, "Woodbine"); + assertEquals("Hello Chad", reader.readLine()); + } + } + + @Test + public void testObjectCodecBytesReplyWithObjectCodecPattern() { + // the object codec only accepts the classes of objectCodecPattern, so a byte[] reply needs [B + Object out = template.requestBody(String.format("mina:tcp://localhost:%d?objectCodecPattern=[B", objectPort), + "Hello"); + assertArrayEquals("Bye Hello".getBytes(StandardCharsets.UTF_8), (byte[]) out); + } + + private static Socket connect(int port) throws Exception { + Socket socket = new Socket(); + socket.connect(new InetSocketAddress("localhost", port)); + socket.setSoTimeout(1000); + return socket; + } + + private static BufferedReader send(Socket socket, String line) throws Exception { + OutputStream os = socket.getOutputStream(); + os.write((line + "\n").getBytes(StandardCharsets.US_ASCII)); + os.flush(); + return new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII)); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + textlinePort = getPort(); + noReplyPort = getNextPort(); + objectPort = getNextPort(); + + fromF("mina:tcp://localhost:%d?textline=true&encoding=US-ASCII", textlinePort) + .choice() + .when(body().isEqualTo("stream")) + .process(e -> e.getMessage().setBody( + new ByteArrayInputStream("Bye stream".getBytes(StandardCharsets.US_ASCII)))) + .when(body().isEqualTo("close")) + .setBody(constant("Bye close")) + .setHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, constant(true)) + .otherwise() + .process(e -> e.getMessage().setBody( + ("Bye " + e.getMessage().getBody(String.class)).getBytes(StandardCharsets.US_ASCII))); + + fromF("mina:tcp://localhost:%d?textline=true&encoding=US-ASCII&disconnectOnNoReply=false&noReplyLogLevel=OFF", + noReplyPort) + .choice() + .when(body().isEqualTo("Woodbine")) + .transform(constant("Hello Chad")) + .otherwise() + .transform(constant((Object) null)); + + fromF("mina:tcp://localhost:%d", objectPort) + .process(e -> e.getMessage().setBody( + ("Bye " + e.getMessage().getBody(String.class)).getBytes(StandardCharsets.UTF_8))); + } + }; + } +} diff --git a/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaSyncFalseTest.java b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaSyncFalseTest.java new file mode 100644 index 000000000000..3132e37bbb63 --- /dev/null +++ b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaSyncFalseTest.java @@ -0,0 +1,113 @@ +/* + * 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.mina; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; + +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.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * With sync=false the endpoints are one-way: the producer keeps the message of an InOut exchange, and the consumer does + * not write a reply. + */ +public class MinaSyncFalseTest extends BaseMinaTest { + + private int udpPort; + + @Test + public void testInOutProducerKeepsBodyAndHeaders() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:tcp"); + mock.expectedBodiesReceived("Hello World"); + + Exchange out = template.request(String.format("mina:tcp://localhost:%d?sync=false&textline=true", getPort()), e -> { + e.getMessage().setBody("Hello World"); + e.getMessage().setHeader("foo", "bar"); + }); + + MockEndpoint.assertIsSatisfied(context); + assertEquals("Hello World", out.getMessage().getBody()); + assertEquals("bar", out.getMessage().getHeader("foo")); + } + + @Test + public void testTcpConsumerDoesNotReply() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:tcp"); + mock.expectedBodiesReceived("Hello World"); + + try (Socket socket = new Socket()) { + socket.connect(new InetSocketAddress("localhost", getPort())); + socket.setSoTimeout(1000); + OutputStream os = socket.getOutputStream(); + os.write("Hello World\n".getBytes(StandardCharsets.US_ASCII)); + os.flush(); + + MockEndpoint.assertIsSatisfied(context); + + // no reply is written back, and the session stays open + BufferedReader reader + = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII)); + assertThrows(SocketTimeoutException.class, reader::readLine); + } + } + + @Test + public void testUdpConsumerDoesNotReply() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:udp"); + mock.expectedBodiesReceived("Hello World"); + + try (DatagramSocket socket = new DatagramSocket()) { + socket.setSoTimeout(1000); + InetAddress address = InetAddress.getByName("127.0.0.1"); + byte[] data = "Hello World".getBytes(StandardCharsets.US_ASCII); + socket.send(new DatagramPacket(data, data.length, address, udpPort)); + + MockEndpoint.assertIsSatisfied(context); + + // no datagram is sent back + byte[] buf = new byte[128]; + DatagramPacket receive = new DatagramPacket(buf, buf.length); + assertThrows(SocketTimeoutException.class, () -> socket.receive(receive)); + } + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + udpPort = getNextPort(); + + fromF("mina:tcp://localhost:%d?sync=false&textline=true", getPort()).to("mock:tcp"); + fromF("mina:udp://127.0.0.1:%d?sync=false", udpPort).convertBodyTo(String.class).to("mock:udp"); + } + }; + } +} 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 10f5a47c1b89..ab7a2900836a 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 @@ -1896,6 +1896,21 @@ does not declare. Before, the asynchronous consumer sent the message of the exce sent a `TApplicationException` as it was, such as one returned by another Thrift service that the route called. An exception declared in the IDL is still sent as it is. Set `muteException=false` to send the exception as before. +=== camel-mina - sync=false is one-way, and the reply of a textline consumer + +A consumer with `sync=false` no longer writes the body the route ends with back to the client. Before, it wrote it back +(over TCP and UDP), so a client that only sends got its own message echoed. Use `sync=true` (the default) when the client +expects a reply. + +A producer with `sync=false` now leaves the body and headers of an `InOut` exchange as they were. Before, the exchange +ended with an empty message. + +A consumer with `textline=true` now converts the reply to a `String` before writing it, as the producer does with the +request. Before, a `byte[]` or a stream reply was written as its `toString()`. + +A consumer with `disconnectOnNoReply=false` now keeps the session open when the route ends with no body, and logs it at +`noReplyLogLevel` (`WARN` by default). Before, the consumer always closed the session and logged at `DEBUG`. + === camel-netty - errors sent to the client A new `muteException` consumer option, enabled by default, makes a consumer with `sync=true` write a generic exception
