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 2d8b6e0da222 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 (#29010)
2d8b6e0da222 is described below
commit 2d8b6e0da222abc654d297ed153d20857cbf3cfc
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 11:41:52 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 (#29010)
* CAMEL-26182: camel-mina - sync=false is one-way: the producer keeps t…
…he 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.
* Merge remote-tracking branch 'origin/main' into CAMEL-26182-mina-sync
# Conflicts:
# docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
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 cbe623761a39..41f803bb6a4a 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
@@ -2053,6 +2053,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-thrift - the reply of the consumer must convert to the return type
of the method
For a method with a return type, the consumer now fails the call with a
`TApplicationException` when the route leaves
