This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26183-mllp-body-conversion in repository https://gitbox.apache.org/repos/asf/camel.git
commit 690e87aa919fc68a60c4708aabf95849e46fd6ac Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:45:46 2026 +0100 CAMEL-26183: camel-mllp - the producer converts a body that is not a String or a byte array The producer handled only a String or a byte[] body. Any other body, such as an InputStream, a StringBuilder or a Map, was sent as an empty MLLP message, and with InOut the producer then timed out waiting for the acknowledgement. A CharSequence is now encoded in the charset like a String, another body is converted to a byte[] with the type converter, and a body that cannot be converted fails fast with a MllpInvalidMessageException. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/components/mllp.json | 2 +- .../apache/camel/catalog/docs/mllp-component.adoc | 5 +- .../org/apache/camel/component/mllp/mllp.json | 2 +- .../camel-mllp/src/main/docs/mllp-component.adoc | 5 +- .../apache/camel/component/mllp/MllpEndpoint.java | 8 +- .../component/mllp/MllpTcpClientProducer.java | 21 +++- .../MllpTcpClientProducerBodyConversionTest.java | 107 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 ++ 8 files changed, 148 insertions(+), 10 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mllp.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mllp.json index 43f0221ec6c1..af65947c1e09 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mllp.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mllp.json @@ -88,7 +88,7 @@ "bodyTypes": { "consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String", "byOption": [ { "option": "stringPayload", "value": "false", "carriers": [ "bytes" ], "javaType": "byte[]" } ], "description": "The HL7 message without the MLLP envelope, as a String in the charset of MSH-18 or charsetName, or a byte array with stringPayload=false" }, "consumerReply": { "carriers": [ "any" ], "description": "The body is not sent back: the acknowledgement is the CamelMllpAcknowledgement or CamelMllpAcknowledgementString exchange property, or with autoAck an HL7 ACK (AA, or AE when the exchange failed)" }, - "producerIn": { "carriers": [ "text", "bytes" ], "description": "The HL7 message as a String or a byte array. No body fails; another body, such as a stream, is sent as an empty message: convert it to a String first" }, + "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ], "description": "The HL7 message as a String (in the charset of charsetName) or a byte array. Another body, such as a stream or a file, is converted to a byte array; no body, or a body that cannot be converted, fails" }, "producerOut": { "carriers": [ "unchanged" ], "description": "The body is left as it was: with InOut the acknowledgement is in the CamelMllpAcknowledgement (byte array) and CamelMllpAcknowledgementString headers" } }, "properties": { diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mllp-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mllp-component.adoc index 383a40fc1504..02bcd48919e5 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mllp-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mllp-component.adoc @@ -32,7 +32,10 @@ The component supports the following: The MLLP consumer gives the HL7 payload as a `String` by default (`stringPayload=true`), or as a `byte[]` with `stringPayload=false`. -The MLLP producer sends a `String` or a `byte[]` body. +The MLLP producer sends a `String` (or another `CharSequence`) or a `byte[]` body, with text encoded in the +`charsetName` charset. Another body, such as an `InputStream` or a file, is converted to a `byte[]` with the +Camel type converter. +A body that cannot be converted fails the exchange with a `MllpInvalidMessageException`. Maven users will need to add the following dependency to their pom.xml for this component: diff --git a/components/camel-mllp/src/generated/resources/META-INF/org/apache/camel/component/mllp/mllp.json b/components/camel-mllp/src/generated/resources/META-INF/org/apache/camel/component/mllp/mllp.json index 43f0221ec6c1..af65947c1e09 100644 --- a/components/camel-mllp/src/generated/resources/META-INF/org/apache/camel/component/mllp/mllp.json +++ b/components/camel-mllp/src/generated/resources/META-INF/org/apache/camel/component/mllp/mllp.json @@ -88,7 +88,7 @@ "bodyTypes": { "consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String", "byOption": [ { "option": "stringPayload", "value": "false", "carriers": [ "bytes" ], "javaType": "byte[]" } ], "description": "The HL7 message without the MLLP envelope, as a String in the charset of MSH-18 or charsetName, or a byte array with stringPayload=false" }, "consumerReply": { "carriers": [ "any" ], "description": "The body is not sent back: the acknowledgement is the CamelMllpAcknowledgement or CamelMllpAcknowledgementString exchange property, or with autoAck an HL7 ACK (AA, or AE when the exchange failed)" }, - "producerIn": { "carriers": [ "text", "bytes" ], "description": "The HL7 message as a String or a byte array. No body fails; another body, such as a stream, is sent as an empty message: convert it to a String first" }, + "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ], "description": "The HL7 message as a String (in the charset of charsetName) or a byte array. Another body, such as a stream or a file, is converted to a byte array; no body, or a body that cannot be converted, fails" }, "producerOut": { "carriers": [ "unchanged" ], "description": "The body is left as it was: with InOut the acknowledgement is in the CamelMllpAcknowledgement (byte array) and CamelMllpAcknowledgementString headers" } }, "properties": { diff --git a/components/camel-mllp/src/main/docs/mllp-component.adoc b/components/camel-mllp/src/main/docs/mllp-component.adoc index 383a40fc1504..02bcd48919e5 100644 --- a/components/camel-mllp/src/main/docs/mllp-component.adoc +++ b/components/camel-mllp/src/main/docs/mllp-component.adoc @@ -32,7 +32,10 @@ The component supports the following: The MLLP consumer gives the HL7 payload as a `String` by default (`stringPayload=true`), or as a `byte[]` with `stringPayload=false`. -The MLLP producer sends a `String` or a `byte[]` body. +The MLLP producer sends a `String` (or another `CharSequence`) or a `byte[]` body, with text encoded in the +`charsetName` charset. Another body, such as an `InputStream` or a file, is converted to a `byte[]` with the +Camel type converter. +A body that cannot be converted fails the exchange with a `MllpInvalidMessageException`. Maven users will need to add the following dependency to their pom.xml for this component: diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java index 6b4ed87a1f59..958c4a182522 100644 --- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java +++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java @@ -65,10 +65,10 @@ import org.slf4j.Logger; + " CamelMllpAcknowledgement or CamelMllpAcknowledgementString exchange" + " property, or with autoAck an HL7 ACK (AA, or AE when the exchange" + " failed)"), - producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES }, - description = "The HL7 message as a String or a byte array. No body fails; another body," - + " such as a stream, is sent as an empty message: convert it to a String" - + " first"), + producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE }, + description = "The HL7 message as a String (in the charset of charsetName) or a byte" + + " array. Another body, such as a stream or a file, is converted to a byte" + + " array; no body, or a body that cannot be converted, fails"), producerOut = @BodyType(value = Carrier.UNCHANGED, description = "The body is left as it was: with InOut the acknowledgement is in the" + " CamelMllpAcknowledgement (byte array) and" diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java index 9eb3c2c69c5c..a01c09087bba 100644 --- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java +++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java @@ -43,6 +43,7 @@ import org.apache.camel.api.management.ManagedResource; import org.apache.camel.component.mllp.internal.Hl7Util; import org.apache.camel.component.mllp.internal.MllpSocketBuffer; import org.apache.camel.support.DefaultProducer; +import org.apache.camel.support.ExchangeHelper; import org.apache.camel.support.jsse.SSLContextParameters; import org.apache.camel.util.StringHelper; import org.slf4j.Logger; @@ -167,12 +168,28 @@ public class MllpTcpClientProducer extends DefaultProducer implements Runnable { return; } else if (messageBody instanceof byte[]) { hl7MessageBytes = (byte[]) messageBody; - } else if (messageBody instanceof String) { - String stringBody = (String) messageBody; + } else if (messageBody instanceof CharSequence) { + String stringBody = messageBody.toString(); hl7MessageBytes = stringBody.getBytes(MllpCharsetHelper.getCharset(exchange, charset)); if (getConfiguration().hasCharsetName()) { exchange.setProperty(ExchangePropertyKey.CHARSET_NAME, getConfiguration().getCharsetName()); } + } else { + // another body, such as a stream: convert it, with text in the same charset as a String body + if (ExchangeHelper.getCharsetName(exchange, false) == null) { + exchange.setProperty(ExchangePropertyKey.CHARSET_NAME, charset.name()); + } + try { + hl7MessageBytes = exchange.getContext().getTypeConverter().mandatoryConvertTo(byte[].class, exchange, + messageBody); + } catch (Exception conversionEx) { + String exceptionMessage = String.format( + "process(%s) [%s] - cannot convert the message body of type %s to a byte array", + exchange.getExchangeId(), socket, messageBody.getClass().getName()); + exchange.setException( + new MllpInvalidMessageException(exceptionMessage, null, conversionEx, logPhi)); + return; + } } log.debug("process({}) [{}] - sending message to external system", exchange.getExchangeId(), socket); diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBodyConversionTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBodyConversionTest.java new file mode 100644 index 000000000000..01c13603feb9 --- /dev/null +++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBodyConversionTest.java @@ -0,0 +1,107 @@ +/* + * 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.mllp; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.EndpointInject; +import org.apache.camel.Exchange; +import org.apache.camel.NoTypeConversionAvailableException; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.AvailablePortFinder; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.apache.camel.test.mllp.Hl7TestMessageGenerator; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The producer converts a body that is not a String or a byte array, instead of sending an empty message. + */ +public class MllpTcpClientProducerBodyConversionTest extends CamelTestSupport { + + @RegisterExtension + AvailablePortFinder.Port mllpPort = AvailablePortFinder.find(); + + @EndpointInject("mock://received") + MockEndpoint received; + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + fromF("mllp://localhost:%d?autoAck=true&stringPayload=false", mllpPort.getPort()) + .to(received); + } + }; + } + + private String producerUri() { + return String.format("mllp://localhost:%d?receiveTimeout=5000&readTimeout=5000&charsetName=UTF-8", + mllpPort.getPort()); + } + + @Test + public void testInputStreamBody() throws Exception { + String testMessage = Hl7TestMessageGenerator.generateMessage(); + byte[] testBytes = testMessage.getBytes(StandardCharsets.ISO_8859_1); + received.expectedMessageCount(1); + + Exchange exchange = template.send(producerUri(), + e -> e.getMessage().setBody(new ByteArrayInputStream(testBytes))); + + MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS); + assertNull(exchange.getException()); + assertArrayEquals(testBytes, received.getExchanges().get(0).getMessage().getBody(byte[].class)); + assertTrue(exchange.getMessage().getHeader(MllpConstants.MLLP_ACKNOWLEDGEMENT_STRING, String.class) + .contains("MSA|AA|00001")); + } + + @Test + public void testTextBodyUsesCharsetName() throws Exception { + String testMessage = Hl7TestMessageGenerator.generateMessage().replace("PID|", "PID|Ærø|"); + received.expectedMessageCount(1); + + Exchange exchange = template.send(producerUri(), + e -> e.getMessage().setBody(new StringBuilder(testMessage))); + + MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS); + assertNull(exchange.getException()); + assertArrayEquals(testMessage.getBytes(StandardCharsets.UTF_8), + received.getExchanges().get(0).getMessage().getBody(byte[].class)); + } + + @Test + public void testBodyThatCannotBeConvertedFails() throws Exception { + received.expectedMessageCount(0); + + Exchange exchange = template.send(producerUri(), e -> e.getMessage().setBody(Map.of("key", "value"))); + + assertInstanceOf(MllpInvalidMessageException.class, exchange.getException()); + assertInstanceOf(NoTypeConversionAvailableException.class, exchange.getException().getCause()); + MockEndpoint.assertIsSatisfied(context); + } +} 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..4663484a45dc 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 @@ -2374,6 +2374,14 @@ connection, and its reconnects run one after the other. The threads of a connect A `MiloClientConnection` created outside an endpoint needs a `CamelContext`, set with `setCamelContext`, before it is used. +=== camel-mllp - the producer converts a body that is not a String or a byte array + +The mllp producer now converts a body that is not a `String` or a `byte[]`, such as an `InputStream` or a file, to a +`byte[]` with the type converter, and encodes a `CharSequence` such as a `StringBuilder` in the `charsetName` +charset. A body that cannot be converted, such as a `Map`, fails the exchange with a `MllpInvalidMessageException`. +Before, such a body was sent as an empty MLLP message: with `InOut` the exchange then failed with a +`MllpAcknowledgementTimeoutException` after the receive timeout, and with `InOnly` it completed without an error. + === camel-lra - only an LRA hosted by the configured coordinator is joined The LRA saga service now joins, completes or compensates an LRA only when its URL is under the configured LRA
