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 394deda18cf8 CAMEL-26433: camel-hl7 - add a maxFrameLength option to
the HL7 MLLP MINA codec (#29043)
394deda18cf8 is described below
commit 394deda18cf817cafe5472cb2dd99428088df612
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 3 16:09:17 2026 +0530
CAMEL-26433: camel-hl7 - add a maxFrameLength option to the HL7 MLLP MINA
codec (#29043)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/hl7-dataformat.adoc | 31 ++-
.../camel-hl7/src/main/docs/hl7-dataformat.adoc | 31 ++-
.../apache/camel/component/hl7/HL7MLLPCodec.java | 14 +
.../apache/camel/component/hl7/HL7MLLPConfig.java | 7 +-
.../apache/camel/component/hl7/HL7MLLPDecoder.java | 141 +++++++++-
.../hl7/HL7MLLPDecoderMaxFrameLengthTest.java | 291 +++++++++++++++++++++
.../hl7/HL7MLLPMinaMaxFrameLengthRouteTest.java | 89 +++++++
7 files changed, 580 insertions(+), 24 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hl7-dataformat.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hl7-dataformat.adoc
index bdbf5edf75b8..5640be46334a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hl7-dataformat.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hl7-dataformat.adoc
@@ -70,14 +70,20 @@ the HL7 message content.
|`convertLFtoCR` |`false` |Will convert `\n` to `\r` (`0x0d`, 13 decimal) as
HL7 stipulates `\r` as
segment terminators. The HAPI library requires the use of `\r`.
-|`maxFrameLength` |`2147483647` |Netty decoder (`HL7MLLPNettyDecoderFactory`)
only: the maximum length in bytes of
-a message, counted from the start of the data up to the end bytes (the start
byte included, the end bytes not). A
-longer message is discarded and fails with a `TooLongFrameException`: the
Netty consumer reports it to its exception
-handler and closes the connection. The decoder fails as soon as more bytes
than the limit have arrived without both end
-bytes, so a message of exactly this length fails when its first end byte
arrives without the second; set the limit
-with some margin over the largest message you expect. The default
(`Integer.MAX_VALUE`) means no limit.
+|`maxFrameLength` |`2147483647` |The maximum length in bytes of a message, for
the MINA codec (`HL7MLLPCodec`) and the
+Netty decoder (`HL7MLLPNettyDecoderFactory`). The default
(`Integer.MAX_VALUE`) means no limit.
|===
+The `maxFrameLength` limit counts from the start of the data up to the end
bytes (the start byte included, the end
+bytes not). Both codecs fail as soon as a message is longer than the limit,
without waiting for its end bytes, and
+discard the rest of the long message up to its end bytes:
+
+* MINA codec: the message fails with a `ProtocolDecoderException`, and the
rest of the data received with it is
+discarded too. The MINA consumer closes the connection, and a MINA producer
fails the exchange when the reply is longer.
+* Netty decoder: the message fails with a `TooLongFrameException`, which the
Netty consumer reports to its exception
+handler before it closes the connection. The Netty decoder counts a first end
byte that arrives without the second, so a
+message of exactly the limit can fail: set the limit with some margin over the
largest message you expect.
+
=== Exposing an HL7 listener using Mina
In the Spring XML file, we configure a mina endpoint to listen for HL7
@@ -103,6 +109,19 @@ up in the Spring XML file:
</bean>
----
+By default the codec accepts messages of any size. To limit the size of a
message, set `maxFrameLength` (in bytes)
+on the codec; a longer message fails with a `ProtocolDecoderException` and the
connection is closed (see
+`maxFrameLength` in the options above):
+
+._XML-only: Spring bean declaration for HL7 MLLP codec with a message size
limit_
+[source,xml]
+----
+<bean id="hl7codec" class="org.apache.camel.component.hl7.HL7MLLPCodec">
+ <property name="charset" value="iso-8859-1"/>
+ <property name="maxFrameLength" value="1048576"/>
+</bean>
+----
+
The endpoint *hl7MinaLlistener* can then be used in a route as a
consumer, as this Java DSL example illustrates:
diff --git a/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
b/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
index bdbf5edf75b8..5640be46334a 100644
--- a/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
+++ b/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
@@ -70,14 +70,20 @@ the HL7 message content.
|`convertLFtoCR` |`false` |Will convert `\n` to `\r` (`0x0d`, 13 decimal) as
HL7 stipulates `\r` as
segment terminators. The HAPI library requires the use of `\r`.
-|`maxFrameLength` |`2147483647` |Netty decoder (`HL7MLLPNettyDecoderFactory`)
only: the maximum length in bytes of
-a message, counted from the start of the data up to the end bytes (the start
byte included, the end bytes not). A
-longer message is discarded and fails with a `TooLongFrameException`: the
Netty consumer reports it to its exception
-handler and closes the connection. The decoder fails as soon as more bytes
than the limit have arrived without both end
-bytes, so a message of exactly this length fails when its first end byte
arrives without the second; set the limit
-with some margin over the largest message you expect. The default
(`Integer.MAX_VALUE`) means no limit.
+|`maxFrameLength` |`2147483647` |The maximum length in bytes of a message, for
the MINA codec (`HL7MLLPCodec`) and the
+Netty decoder (`HL7MLLPNettyDecoderFactory`). The default
(`Integer.MAX_VALUE`) means no limit.
|===
+The `maxFrameLength` limit counts from the start of the data up to the end
bytes (the start byte included, the end
+bytes not). Both codecs fail as soon as a message is longer than the limit,
without waiting for its end bytes, and
+discard the rest of the long message up to its end bytes:
+
+* MINA codec: the message fails with a `ProtocolDecoderException`, and the
rest of the data received with it is
+discarded too. The MINA consumer closes the connection, and a MINA producer
fails the exchange when the reply is longer.
+* Netty decoder: the message fails with a `TooLongFrameException`, which the
Netty consumer reports to its exception
+handler before it closes the connection. The Netty decoder counts a first end
byte that arrives without the second, so a
+message of exactly the limit can fail: set the limit with some margin over the
largest message you expect.
+
=== Exposing an HL7 listener using Mina
In the Spring XML file, we configure a mina endpoint to listen for HL7
@@ -103,6 +109,19 @@ up in the Spring XML file:
</bean>
----
+By default the codec accepts messages of any size. To limit the size of a
message, set `maxFrameLength` (in bytes)
+on the codec; a longer message fails with a `ProtocolDecoderException` and the
connection is closed (see
+`maxFrameLength` in the options above):
+
+._XML-only: Spring bean declaration for HL7 MLLP codec with a message size
limit_
+[source,xml]
+----
+<bean id="hl7codec" class="org.apache.camel.component.hl7.HL7MLLPCodec">
+ <property name="charset" value="iso-8859-1"/>
+ <property name="maxFrameLength" value="1048576"/>
+</bean>
+----
+
The endpoint *hl7MinaLlistener* can then be used in a route as a
consumer, as this Java DSL example illustrates:
diff --git
a/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPCodec.java
b/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPCodec.java
index 237d07b392e6..2d67636dbd54 100644
---
a/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPCodec.java
+++
b/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPCodec.java
@@ -118,4 +118,18 @@ public class HL7MLLPCodec implements ProtocolCodecFactory {
public void setProduceString(boolean apply) {
config.setProduceString(apply);
}
+
+ public int getMaxFrameLength() {
+ return config.getMaxFrameLength();
+ }
+
+ /**
+ * The maximum length in bytes of a message that the decoder reads,
counted from the start of the data up to the end
+ * bytes (the start byte included, the end bytes not). A longer message is
discarded, with the rest of the data
+ * received with it, and fails with a {@code ProtocolDecoderException}; a
MINA consumer closes the connection, and a
+ * MINA producer fails the exchange when a reply is longer. The default is
{@link Integer#MAX_VALUE} (no limit).
+ */
+ public void setMaxFrameLength(int maxFrameLength) {
+ config.setMaxFrameLength(maxFrameLength);
+ }
}
diff --git
a/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPConfig.java
b/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPConfig.java
index 8e7a0aed356b..79159a2940fa 100644
---
a/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPConfig.java
+++
b/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPConfig.java
@@ -142,9 +142,10 @@ public class HL7MLLPConfig {
}
/**
- * The maximum length in bytes of a message that the Netty decoder reads,
counted from the start of the data up to
- * the end bytes (the start byte included, the end bytes not). A longer
message is discarded and fails with a
- * {@code TooLongFrameException}. The default is {@link Integer#MAX_VALUE}
(no limit).
+ * The maximum length in bytes of a message that the decoder (Netty or
MINA) reads, counted from the start of the
+ * data up to the end bytes (the start byte included, the end bytes not).
A longer message is discarded and fails
+ * with a {@code TooLongFrameException} (Netty) or a {@code
ProtocolDecoderException} (MINA). The default is
+ * {@link Integer#MAX_VALUE} (no limit).
*/
public void setMaxFrameLength(int maxFrameLength) {
if (maxFrameLength <= 0) {
diff --git
a/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPDecoder.java
b/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPDecoder.java
index 842ce5030a38..a3dfc7844b5b 100644
---
a/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPDecoder.java
+++
b/components/camel-hl7/src/main/java/org/apache/camel/component/hl7/HL7MLLPDecoder.java
@@ -22,6 +22,7 @@ import java.nio.charset.CharsetDecoder;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
+import org.apache.mina.filter.codec.ProtocolDecoderException;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,6 +41,10 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
HL7MLLPDecoder(HL7MLLPConfig config) {
this.config = config;
+ // keep the data of an incomplete message between two reads also on a
transport that MINA does not regard as
+ // fragmenting (mina:udp, mina:vm): without it MINA hands each read to
doDecode on its own, so the positions
+ // that this decoder keeps would point into a read that MINA did not
keep
+ setTransportMetadataFragmentation(false);
}
@Override
@@ -50,13 +55,22 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
DecoderState state = decoderState(session);
in.position(state.current());
+ if (state.isDiscarding() && !skipDiscarded(state, in)) {
+ // the rest of a message that was too long: drop what was
received, and wait for its end bytes
+ LOG.debug("Discarded {} bytes of a message longer than
maxFrameLength", in.limit() - state.current());
+ in.position(in.limit());
+ state.markCurrent(0);
+ return false;
+ }
+
LOG.debug("Received data, checking from position {} to {}",
in.position(), in.limit());
boolean messageDecoded = false;
while (in.hasRemaining()) {
int previousPosition = in.position();
- byte current = in.get();
+ // unsigned, to compare it with the configured start and end bytes
(chars), also above 0x7F
+ int current = in.get() & 0xFF;
// Check if we are at the end of an HL7 message
if (current == config.getEndByte2() && state.previous() ==
config.getEndByte1()) {
@@ -81,11 +95,22 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
in.position(currentPosition);
in.limit(currentLimit);
state.reset();
+ // the data of the next message starts after these end
bytes
+ state.markDataStart(currentPosition);
}
} else {
LOG.warn("Ignoring message end at position {} until start
byte has been seen.", previousPosition);
+ // the data of the next message starts after these end
bytes
+ state.markDataStart(in.position());
}
} else {
+ // the bytes of the message so far, from the start of its
data; an end byte 1 is not counted yet, as the
+ // end byte 2 may follow it
+ int length = in.position() - state.dataStart() - (current ==
config.getEndByte1() ? 1 : 0);
+ if (length > config.getMaxFrameLength()) {
+ state.markPrevious(current);
+ tooLong(session, in, state);
+ }
// Check if we are at the start of an HL7 message
if (current == config.getStartByte()) {
state.markStart(previousPosition);
@@ -99,16 +124,81 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
}
if (!messageDecoded) {
- // Could not find a complete message in the buffer.
- // Reset to the initial position (just as nothing had been read
yet)
- // and return false so that this method is called again with more
data.
+ // Could not find a complete message in the buffer: return false
so that this method is called again
+ // with more data. Keep only the data of the current message: MINA
keeps the bytes from this position for
+ // the next call (so the bytes of the messages decoded before, and
of end bytes without a start, do not
+ // pile up), and the positions of the state are moved down by as
much
LOG.debug("No complete message yet at position {}", in.position());
- state.markCurrent(in.position());
- in.position(0);
+ int dataStart = state.dataStart();
+ state.markCurrent(in.position() - dataStart);
+ state.rebase(dataStart);
+ in.position(dataStart);
+ } else {
+ // the buffer is used up to the end of this message, so the next
data starts at the start of the buffer
+ state.markDataStart(0);
}
return messageDecoded;
}
+ /**
+ * Skips the bytes of a message that was too long up to its end bytes.
+ *
+ * @return true when its end bytes were found, and the data of the next
message starts after them
+ */
+ private boolean skipDiscarded(DecoderState state, IoBuffer in) {
+ while (in.hasRemaining()) {
+ int current = in.get() & 0xFF;
+ if (current == config.getEndByte2() && state.previous() ==
config.getEndByte1()) {
+ LOG.debug("End of the discarded message at position {}",
in.position() - 1);
+ state.stopDiscarding();
+ state.markPrevious(0);
+ state.markDataStart(in.position());
+ return true;
+ }
+ state.markPrevious(current);
+ }
+ return false;
+ }
+
+ /**
+ * Discards a message longer than maxFrameLength, with the rest of the
data received with it, and fails.
+ */
+ private void tooLong(IoSession session, IoBuffer in, DecoderState state)
throws Exception {
+ // the end bytes of this message may already be in the rest of the
data: then the next message starts after them
+ boolean discarding = !skipDiscarded(state, in);
+ int previous = state.previous();
+ in.position(in.limit());
+ state.reset();
+ state.markDataStart(0);
+ state.markPrevious(previous);
+ if (discarding) {
+ state.startDiscarding();
+ }
+ // the data is used up: drop the buffer that MINA keeps for this
session, so that it is not read again (this
+ // decoder throws, so MINA does not do it)
+ super.dispose(session);
+ throw new FrameTooLongException(
+ "HL7 MLLP message is longer than maxFrameLength of " +
config.getMaxFrameLength() + " bytes: discarded");
+ }
+
+ /**
+ * The failure of a message longer than maxFrameLength. It has an empty
hex dump, so that MINA does not add the data
+ * that was received (the message) to it, and so to the logs
(ProtocolCodecFilter dumps the data only when the
+ * exception has none).
+ */
+ static final class FrameTooLongException extends ProtocolDecoderException {
+ private static final long serialVersionUID = 1L;
+
+ FrameTooLongException(String message) {
+ super(message);
+ }
+
+ @Override
+ public String getHexdump() {
+ return "";
+ }
+ }
+
// Make a defensive byte copy (the buffer will be reused)
// and omit the start and the two end bytes of the MLLP message
// returning a byte array
@@ -186,7 +276,12 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
private static class DecoderState {
private int startPos = -1;
private int currentPos;
- private byte previousByte;
+ // unsigned (0 to 0xFF)
+ private int previousByte;
+ // where the data of the current message starts in the buffer: after
the end bytes of the previous message
+ private int dataStartPos;
+ // whether the rest of a message that was too long is being skipped
+ private boolean discarding;
void reset() {
startPos = -1;
@@ -207,7 +302,35 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
currentPos = position;
}
- void markPrevious(byte previous) {
+ void markDataStart(int position) {
+ dataStartPos = position;
+ }
+
+ /** Moves the positions down by {@code offset}, as the first {@code
offset} bytes of the buffer are dropped. */
+ void rebase(int offset) {
+ if (startPos >= 0) {
+ startPos -= offset;
+ }
+ dataStartPos -= offset;
+ }
+
+ int dataStart() {
+ return dataStartPos;
+ }
+
+ void startDiscarding() {
+ discarding = true;
+ }
+
+ void stopDiscarding() {
+ discarding = false;
+ }
+
+ boolean isDiscarding() {
+ return discarding;
+ }
+
+ void markPrevious(int previous) {
previousByte = previous;
}
@@ -219,7 +342,7 @@ class HL7MLLPDecoder extends CumulativeProtocolDecoder {
return currentPos;
}
- public byte previous() {
+ public int previous() {
return previousByte;
}
diff --git
a/components/camel-hl7/src/test/java/org/apache/camel/component/hl7/HL7MLLPDecoderMaxFrameLengthTest.java
b/components/camel-hl7/src/test/java/org/apache/camel/component/hl7/HL7MLLPDecoderMaxFrameLengthTest.java
new file mode 100644
index 000000000000..515b4fa38098
--- /dev/null
+++
b/components/camel-hl7/src/test/java/org/apache/camel/component/hl7/HL7MLLPDecoderMaxFrameLengthTest.java
@@ -0,0 +1,291 @@
+/*
+ * 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.hl7;
+
+import java.lang.reflect.Proxy;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.mina.core.buffer.IoBuffer;
+import org.apache.mina.core.filterchain.IoFilter;
+import org.apache.mina.filter.codec.ProtocolCodecFilter;
+import org.apache.mina.filter.codec.ProtocolCodecSession;
+import org.apache.mina.filter.codec.ProtocolDecoder;
+import org.apache.mina.filter.codec.ProtocolDecoderException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests the maxFrameLength option of the HL7 MLLP MINA decoder. The decoder
keeps the data that has no complete message
+ * yet between two calls, on a TCP session and also on a session without
fragmentation, such as the test session.
+ */
+class HL7MLLPDecoderMaxFrameLengthTest {
+
+ private static final String START = "\u000b";
+ private static final String END = "\u001c\r";
+
+ private final ProtocolCodecSession session = new ProtocolCodecSession();
+
+ @Test
+ void messageAtTheLimitIsDecoded() throws Exception {
+ // the start byte and 49 bytes of message: 50 bytes before the end
bytes
+ String message = "A".repeat(49);
+ ProtocolDecoder decoder = decoder(50);
+
+ decode(decoder, START + message + END);
+
+ assertEquals(message, session.getDecoderOutputQueue().poll());
+ assertNull(session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void messageAtTheLimitIsDecodedWhenItsEndBytesArriveApart() throws
Exception {
+ // unlike the Netty decoder, an end byte 1 that arrives without the
end byte 2 is not counted yet
+ String message = "A".repeat(49);
+ ProtocolDecoder decoder = decoder(50);
+
+ decode(decoder, START + message + "\u001c");
+ decode(decoder, "\r");
+
+ assertEquals(message, session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void messageOverTheLimitFails() throws Exception {
+ ProtocolDecoder decoder = decoder(50);
+
+ ProtocolDecoderException e =
assertThrows(ProtocolDecoderException.class,
+ () -> decode(decoder, START + "A".repeat(50) + END));
+
+ assertTrue(e.getMessage().startsWith("HL7 MLLP message is longer than
maxFrameLength of 50 bytes"),
+ e.getMessage());
+ // MINA adds a hex dump of the received data (the message) to a
decoder exception that has none, and the
+ // consumer logs the exception: this one has an empty dump
+ assertEquals("", e.getHexdump());
+ assertFalse(e.getMessage().contains("Hexdump"), e.getMessage());
+ assertNull(session.getDecoderOutputQueue().poll());
+
+ // its end bytes came with it: the next message is decoded
+ decode(decoder, START + "next" + END);
+ assertEquals("next", session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void messageOverTheLimitOverTwoReadsFailsAndTheNextMessageIsDecoded()
throws Exception {
+ ProtocolDecoder decoder = decoder(50);
+
+ // the first part is kept for the next read; the second part goes over
the limit and has the end bytes
+ decode(decoder, START + "B".repeat(30));
+ assertThrows(ProtocolDecoderException.class, () -> decode(decoder,
"B".repeat(30) + END));
+ assertNull(session.getDecoderOutputQueue().poll());
+
+ // nothing of the long message is read again: the next message is
decoded
+ decode(decoder, START + "next" + END);
+ assertEquals("next", session.getDecoderOutputQueue().poll());
+ assertNull(session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void dataWithoutEndBytesFailsOnceOverTheLimitAndTheNextMessageIsDecoded()
throws Exception {
+ ProtocolDecoder decoder = decoder(50);
+
+ // a peer that sends more than the limit without the end bytes: the
decoder does not wait for them, in one read
+ // or over several reads
+ decode(decoder, START + "B".repeat(30));
+ assertThrows(ProtocolDecoderException.class, () -> decode(decoder,
"B".repeat(30)));
+ assertNull(session.getDecoderOutputQueue().poll());
+
+ // the rest of the long message is discarded up to its end bytes (over
several reads too), and the next message
+ // is decoded
+ decode(decoder, "B".repeat(100));
+ decode(decoder, "B".repeat(20) + "\u001c");
+ decode(decoder, "\r" + START + "next" + END);
+ assertEquals("next", session.getDecoderOutputQueue().poll());
+ assertNull(session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void dataBeforeTheStartByteCounts() throws Exception {
+ // the limit counts from the start of the data, as the Netty decoder
does: bytes before the start byte count too
+ ProtocolDecoder decoder = decoder(50);
+
+ assertThrows(ProtocolDecoderException.class, () -> decode(decoder,
"x".repeat(45) + START + "A".repeat(10) + END));
+ decode(decoder, START + "next" + END);
+ assertEquals("next", session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void eachMessageOfOneReadHasTheLimit() throws Exception {
+ // the limit counts from the end bytes of the previous message, also
when several messages come in one read
+ ProtocolDecoder decoder = decoder(50);
+ String message = "A".repeat(49);
+
+ decode(decoder, START + message + END + START + message + END + START
+ message.substring(10));
+ decode(decoder, "A".repeat(10) + END);
+
+ assertEquals(message, session.getDecoderOutputQueue().poll());
+ assertEquals(message, session.getDecoderOutputQueue().poll());
+ assertEquals(message, session.getDecoderOutputQueue().poll());
+ assertNull(session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void theDataOfDecodedMessagesIsNotKept() throws Exception {
+ // reads that do not end with a message end: only the data of the
current message is kept for the next read
+ ProtocolDecoder decoder = decoder(50);
+ for (int i = 0; i < 1000; i++) {
+ decode(decoder, "A" + END + START + "message-" + i);
+ decode(decoder, END + START + "next-" + i);
+ }
+ assertTrue(sessionBufferSize() <= 50, "the session buffer has " +
sessionBufferSize() + " bytes");
+ // each read ends the message of the previous one (the first "A" has
no start byte), the last one stays open
+ assertEquals(1999, session.getDecoderOutputQueue().size());
+ assertEquals("message-0", session.getDecoderOutputQueue().poll());
+ assertEquals("next-0A", session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void endBytesWithoutAStartAreNotKept() throws Exception {
+ // end bytes without a start byte are ignored (with a WARN), and the
data before them is not kept
+ ProtocolDecoder decoder = decoder(50);
+ for (int i = 0; i < 100; i++) {
+ decode(decoder, "x" + END + "x");
+ }
+ assertTrue(sessionBufferSize() <= 50, "the session buffer has " +
sessionBufferSize() + " bytes");
+ decode(decoder, END + START + "next" + END);
+ assertEquals("next", session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ // without the data kept, MINA's codec filter loops on the read for good:
fail instead
+ @Timeout(value = 10, unit = TimeUnit.SECONDS, threadMode =
Timeout.ThreadMode.SEPARATE_THREAD)
+ void aTransportWithoutFragmentationKeepsTheDataOfAnIncompleteMessage()
throws Exception {
+ // the test session reports no fragmentation, as a mina:udp or mina:vm
session: through the codec filter of
+ // MINA, a read with a message and the start of the next one gives
each message once
+ ProtocolCodecFilter filter = new ProtocolCodecFilter(new
HL7MLLPCodec());
+ List<Object> received = new ArrayList<>();
+ IoFilter.NextFilter next = (IoFilter.NextFilter)
Proxy.newProxyInstance(getClass().getClassLoader(),
+ new Class<?>[] { IoFilter.NextFilter.class }, (proxy, method,
args) -> {
+ if ("messageReceived".equals(method.getName()) ||
"exceptionCaught".equals(method.getName())) {
+ received.add(args[1]);
+ if (received.size() > 2) {
+ // the same message again: stop the loop of the
filter instead of filling the memory
+ throw new IllegalStateException("received " +
received);
+ }
+ }
+ return null;
+ });
+
+ filter.messageReceived(next, session,
+ IoBuffer.wrap((START + "A" + END + START +
"B").getBytes(StandardCharsets.ISO_8859_1)));
+ filter.messageReceived(next, session,
IoBuffer.wrap(END.getBytes(StandardCharsets.ISO_8859_1)));
+
+ assertEquals(List.of("A", "B"), received);
+ }
+
+ @Test
+ void startAndEndBytesAbove0x7F() throws Exception {
+ // the decoder compares the bytes it reads with the configured start
and end bytes unsigned
+ HL7MLLPConfig config = new HL7MLLPConfig();
+ config.setCharset(StandardCharsets.ISO_8859_1);
+ config.setStartByte((char) 0x8b);
+ config.setEndByte1((char) 0x9c);
+ config.setEndByte2((char) 0x8d);
+ config.setMaxFrameLength(50);
+ ProtocolDecoder decoder = new HL7MLLPDecoder(config);
+ String start = "\u008b";
+ String end = "\u009c\u008d";
+
+ decode(decoder, start + "A".repeat(49) + end);
+ assertEquals("A".repeat(49), session.getDecoderOutputQueue().poll());
+
+ // an end byte 1 that arrives without the end byte 2 is not counted yet
+ decode(decoder, start + "B".repeat(49) + "\u009c");
+ decode(decoder, "\u008d");
+ assertEquals("B".repeat(49), session.getDecoderOutputQueue().poll());
+
+ // a message over the limit is discarded up to its end bytes, and the
next message is decoded
+ assertThrows(ProtocolDecoderException.class, () -> decode(decoder,
start + "C".repeat(60)));
+ decode(decoder, "C".repeat(10) + end + start + "next" + end);
+ assertEquals("next", session.getDecoderOutputQueue().poll());
+ assertNull(session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void codecAppliesTheLimitToItsDecoders() throws Exception {
+ HL7MLLPCodec codec = new HL7MLLPCodec();
+ codec.setCharset(StandardCharsets.ISO_8859_1);
+ codec.setMaxFrameLength(50);
+ assertEquals(50, codec.getMaxFrameLength());
+ ProtocolDecoder decoder = codec.getDecoder(session);
+
+ decode(decoder, START + "A".repeat(49) + END);
+ assertEquals("A".repeat(49), session.getDecoderOutputQueue().poll());
+
+ assertThrows(ProtocolDecoderException.class, () -> decode(decoder,
START + "A".repeat(50) + END));
+ assertNull(session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void defaultIsNoLimit() throws Exception {
+ assertEquals(Integer.MAX_VALUE, new
HL7MLLPCodec().getMaxFrameLength());
+
+ String message = "C".repeat(1024 * 1024);
+ HL7MLLPCodec codec = new HL7MLLPCodec();
+ codec.setCharset(StandardCharsets.ISO_8859_1);
+ ProtocolDecoder decoder = codec.getDecoder(session);
+ decode(decoder, START + message.substring(1000));
+ decode(decoder, message.substring(0, 1000) + END);
+
+ assertEquals(message, session.getDecoderOutputQueue().poll());
+ }
+
+ @Test
+ void maxFrameLengthMustBePositive() {
+ HL7MLLPCodec codec = new HL7MLLPCodec();
+ assertThrows(IllegalArgumentException.class, () ->
codec.setMaxFrameLength(0));
+ assertThrows(IllegalArgumentException.class, () ->
codec.setMaxFrameLength(-1));
+ assertEquals(Integer.MAX_VALUE, codec.getMaxFrameLength());
+ }
+
+ /**
+ * The bytes that MINA keeps for the session between two reads (0 when it
keeps none).
+ */
+ private int sessionBufferSize() {
+ return
session.getAttributeKeys().stream().map(session::getAttribute).filter(IoBuffer.class::isInstance)
+ .mapToInt(b -> ((IoBuffer) b).position()).sum();
+ }
+
+ private void decode(ProtocolDecoder decoder, String data) throws Exception
{
+ decoder.decode(session,
IoBuffer.wrap(data.getBytes(StandardCharsets.ISO_8859_1)),
session.getDecoderOutput());
+ }
+
+ private static ProtocolDecoder decoder(int maxFrameLength) {
+ HL7MLLPConfig config = new HL7MLLPConfig();
+ config.setCharset(StandardCharsets.ISO_8859_1);
+ config.setMaxFrameLength(maxFrameLength);
+ return new HL7MLLPDecoder(config);
+ }
+}
diff --git
a/components/camel-hl7/src/test/java/org/apache/camel/component/hl7/HL7MLLPMinaMaxFrameLengthRouteTest.java
b/components/camel-hl7/src/test/java/org/apache/camel/component/hl7/HL7MLLPMinaMaxFrameLengthRouteTest.java
new file mode 100644
index 000000000000..ea08eeb4e9f1
--- /dev/null
+++
b/components/camel-hl7/src/test/java/org/apache/camel/component/hl7/HL7MLLPMinaMaxFrameLengthRouteTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.hl7;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketException;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.camel.BindToRegistry;
+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;
+
+/**
+ * A MINA HL7 listener with a maxFrameLength refuses a longer message and
keeps serving other connections.
+ */
+class HL7MLLPMinaMaxFrameLengthRouteTest extends HL7TestSupport {
+
+ private static final String MESSAGE =
"MSH|^~\\&|MYSENDER|MYRECEIVER|MYAPPLICATION||200612211200||QRY^A19|1234|P|2.4";
+
+ @BindToRegistry("hl7codec")
+ HL7MLLPCodec addCodec() {
+ HL7MLLPCodec codec = new HL7MLLPCodec();
+ codec.setCharset("iso-8859-1");
+ codec.setMaxFrameLength(1024);
+ return codec;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("mina:tcp://127.0.0.1:" + getPort() +
"?sync=true&codec=#hl7codec")
+ .to("mock:result")
+ .setBody(constant("ACK"));
+ }
+ };
+ }
+
+ @Test
+ void messageOverTheLimitClosesItsConnectionAndOtherMessagesAreServed()
throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived(MESSAGE);
+
+ try (Socket socket = new Socket("127.0.0.1", getPort())) {
+ socket.setSoTimeout(10000);
+ OutputStream out = socket.getOutputStream();
+ // a message of 2000 bytes without the end bytes: the listener
does not wait for them
+ out.write(0x0b);
+ out.write("A".repeat(2000).getBytes(StandardCharsets.ISO_8859_1));
+ out.flush();
+ assertEquals(-1, readUntilClosed(socket.getInputStream()));
+ }
+
+ String reply = template.requestBody("mina:tcp://127.0.0.1:" +
getPort() + "?sync=true&codec=#hl7codec", MESSAGE,
+ String.class);
+ assertEquals("ACK", reply);
+
+ mock.assertIsSatisfied();
+ }
+
+ private static int readUntilClosed(InputStream in) throws IOException {
+ try {
+ return in.read();
+ } catch (SocketException e) {
+ // connection reset by the listener
+ return -1;
+ }
+ }
+}
