This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 6b3ba507e012 [backport 4.22.x] CAMEL-24668: Fix MLLP consumer
discarding pipelined HL7 messages (#26261)
6b3ba507e012 is described below
commit 6b3ba507e012eb04e636756fef51810fcf4a4f98
Author: JinyuChen97 <[email protected]>
AuthorDate: Thu Sep 10 15:21:10 2026 +0100
[backport 4.22.x] CAMEL-24668: Fix MLLP consumer discarding pipelined HL7
messages (#26261)
Pipelined MLLP messages (multiple HL7 messages sent in a single TCP write)
were silently dropped because the buffer only tracked the first envelope.
Extract a processBufferedMessages() loop that preserves trailing data across
reset/ACK cycles, and add isCompleteEnvelopeAt() to distinguish real
pipelined
messages from junk containing SOB bytes. Also handles senders that terminate
frames with <FS><CR><LF>, where the trailing <LF> previously discarded the
next pipelined message when it spanned multiple reads.
Co-authored-by: Federico Mariani <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>
Closes #26261
---
.../component/mllp/internal/MllpSocketBuffer.java | 27 +++++
.../mllp/internal/TcpSocketConsumerRunnable.java | 108 +++++++++++++----
...MllpTcpServerConsumerPipelinedMessagesTest.java | 133 +++++++++++++++++++++
.../test/junit/rule/mllp/MllpClientResource.java | 54 +++++++++
4 files changed, 300 insertions(+), 22 deletions(-)
diff --git
a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/MllpSocketBuffer.java
b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/MllpSocketBuffer.java
index ca8bb67a2589..eb0a8bcb1e43 100644
---
a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/MllpSocketBuffer.java
+++
b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/MllpSocketBuffer.java
@@ -553,6 +553,33 @@ public class MllpSocketBuffer {
}
}
+ /**
+ * Determine whether a complete MLLP envelope begins at the supplied
buffer position.
+ *
+ * @param startIndex the position expected to contain {@link
MllpProtocolConstants#START_OF_BLOCK}
+ * @return {@code true} if a complete envelope begins at {@code
startIndex}
+ */
+ public boolean isCompleteEnvelopeAt(int startIndex) {
+ lock.lock();
+ try {
+ if (startIndex < 0 || startIndex >= availableByteCount
+ || buffer[startIndex] !=
MllpProtocolConstants.START_OF_BLOCK) {
+ return false;
+ }
+
+ for (int i = startIndex + 1; i < availableByteCount; i++) {
+ if (buffer[i] == MllpProtocolConstants.END_OF_BLOCK) {
+ return !isEndOfDataRequired()
+ || i + 1 < availableByteCount && buffer[i + 1] ==
MllpProtocolConstants.END_OF_DATA;
+ }
+ }
+
+ return false;
+ } finally {
+ lock.unlock();
+ }
+ }
+
public boolean hasStartOfBlock() {
lock.lock();
try {
diff --git
a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/TcpSocketConsumerRunnable.java
b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/TcpSocketConsumerRunnable.java
index 794371fab455..db05e0120398 100644
---
a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/TcpSocketConsumerRunnable.java
+++
b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/TcpSocketConsumerRunnable.java
@@ -20,8 +20,10 @@ import java.io.IOException;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
+import java.util.Arrays;
import org.apache.camel.Route;
+import org.apache.camel.component.mllp.MllpProtocolConstants;
import org.apache.camel.component.mllp.MllpSocketException;
import org.apache.camel.component.mllp.MllpTcpServerConsumer;
import org.apache.camel.spi.UnitOfWork;
@@ -135,12 +137,9 @@ public class TcpSocketConsumerRunnable implements Runnable
{
log.debug("Starting {} for {}", this.getClass().getSimpleName(),
combinedAddress);
try {
- byte[] hl7MessageBytes = null;
if (mllpBuffer.hasCompleteEnvelope()) {
- // If we got a complete message on the validation read,
process it
- hl7MessageBytes = mllpBuffer.toMllpPayload();
- mllpBuffer.reset();
- consumer.processMessage(hl7MessageBytes, this);
+ // Process all complete messages received during the
validation read.
+ processBufferedMessages();
}
while (running && null != clientSocket &&
clientSocket.isConnected() && !clientSocket.isClosed()) {
@@ -148,23 +147,7 @@ public class TcpSocketConsumerRunnable implements Runnable
{
try {
mllpBuffer.readFrom(clientSocket);
if (mllpBuffer.hasCompleteEnvelope()) {
- hl7MessageBytes = mllpBuffer.toMllpPayload();
- if (log.isDebugEnabled()) {
- log.debug("Received {} byte message {}",
hl7MessageBytes.length,
-
hl7Util.convertToLoggableString(hl7MessageBytes));
- }
- if (mllpBuffer.hasLeadingOutOfBandData()) {
- // TODO: Move the conversion utilities to the
MllpSocketBuffer to avoid a byte[] copy
- log.warn("Ignoring leading out-of-band data: {}",
-
hl7Util.convertToLoggableString(mllpBuffer.getLeadingOutOfBandData()));
- }
- if (mllpBuffer.hasTrailingOutOfBandData()) {
- log.warn("Ignoring trailing out-of-band data: {}",
-
hl7Util.convertToLoggableString(mllpBuffer.getTrailingOutOfBandData()));
- }
- mllpBuffer.reset();
-
- consumer.processMessage(hl7MessageBytes, this);
+ processBufferedMessages();
} else if (!mllpBuffer.hasStartOfBlock()) {
byte[] payload = mllpBuffer.toByteArray();
log.warn("Ignoring {} byte un-enveloped payload {}",
payload.length,
@@ -217,6 +200,87 @@ public class TcpSocketConsumerRunnable implements Runnable
{
}
}
+ /**
+ * Process all complete MLLP envelopes currently in the buffer.
+ * <p>
+ * The buffer is reused to generate and send the acknowledgement. Preserve
a trailing framed message before
+ * processing the current one and restore it afterwards, so pipelined
messages are not lost when the buffer is reset
+ * for the acknowledgement.
+ */
+ private void processBufferedMessages() {
+ do {
+ byte[] hl7MessageBytes = mllpBuffer.toMllpPayload();
+ if (log.isDebugEnabled()) {
+ log.debug("Received {} byte message {}",
hl7MessageBytes.length,
+ hl7Util.convertToLoggableString(hl7MessageBytes));
+ }
+ if (mllpBuffer.hasLeadingOutOfBandData()) {
+ log.warn("Ignoring leading out-of-band data: {}",
+
hl7Util.convertToLoggableString(mllpBuffer.getLeadingOutOfBandData()));
+ }
+
+ byte[] trailingMessageData = extractTrailingMessageData();
+ mllpBuffer.reset();
+ consumer.processMessage(hl7MessageBytes, this);
+
+ if (trailingMessageData != null) {
+ mllpBuffer.reset();
+ mllpBuffer.write(trailingMessageData);
+ }
+ } while (isSocketOpen() && mllpBuffer.hasCompleteEnvelope());
+
+ if (!mllpBuffer.isEmpty() && !isSocketOpen()) {
+ log.warn("Abandoning {} bytes of unprocessed pipelined data
because the connection is closed",
+ mllpBuffer.size());
+ }
+ }
+
+ /**
+ * Return trailing data beginning with the next START_OF_BLOCK, if
present. Data preceding it, and trailing data
+ * with no START_OF_BLOCK, is out-of-band data.
+ */
+ private byte[] extractTrailingMessageData() {
+ if (!mllpBuffer.hasTrailingOutOfBandData()) {
+ return null;
+ }
+
+ byte[] trailingData = mllpBuffer.getTrailingOutOfBandData();
+ int trailingDataOffset = mllpBuffer.size() - trailingData.length;
+ int leadingLineBreakCount = 0;
+ while (leadingLineBreakCount < trailingData.length &&
isLineBreak(trailingData[leadingLineBreakCount])) {
+ leadingLineBreakCount++;
+ }
+ for (int i = 0; i < trailingData.length; i++) {
+ if (trailingData[i] == MllpProtocolConstants.START_OF_BLOCK) {
+ // A START_OF_BLOCK preceded only by line breaks (e.g. senders
terminating frames with <FS><CR><LF>)
+ // starts the next message, even if that message is not
complete yet.
+ if (i == leadingLineBreakCount ||
mllpBuffer.isCompleteEnvelopeAt(trailingDataOffset + i)) {
+ if (i > 0) {
+ byte[] outOfBandData = Arrays.copyOf(trailingData, i);
+ log.warn("Ignoring {} bytes of out-of-band data before
next message: {}", i,
+
hl7Util.convertToLoggableString(outOfBandData));
+ }
+ return Arrays.copyOfRange(trailingData, i,
trailingData.length);
+ }
+
+ // A START_OF_BLOCK embedded in junk must not be retained as a
partial message: doing so would
+ // turn a benign trailing-data warning into a receive-timeout
error and reset the connection.
+ continue;
+ }
+ }
+
+ log.warn("Ignoring trailing out-of-band data: {}",
hl7Util.convertToLoggableString(trailingData));
+ return null;
+ }
+
+ private static boolean isLineBreak(byte b) {
+ return b == '\r' || b == '\n';
+ }
+
+ private boolean isSocketOpen() {
+ return running && clientSocket != null && clientSocket.isConnected()
&& !clientSocket.isClosed();
+ }
+
public Socket getSocket() {
return clientSocket;
}
diff --git
a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerPipelinedMessagesTest.java
b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerPipelinedMessagesTest.java
new file mode 100644
index 000000000000..4a37f0d4d673
--- /dev/null
+++
b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerPipelinedMessagesTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.util.concurrent.TimeUnit;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit.rule.mllp.MllpClientResource;
+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.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class MllpTcpServerConsumerPipelinedMessagesTest extends
CamelTestSupport {
+ @RegisterExtension
+ public MllpClientResource mllpClient = new MllpClientResource();
+
+ @EndpointInject("mock://result")
+ MockEndpoint result;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ mllpClient.setMllpHost("localhost");
+ mllpClient.setMllpPort(AvailablePortFinder.getNextAvailable());
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("mllp://%s:%d?minBufferSize=8192",
mllpClient.getMllpHost(), mllpClient.getMllpPort())
+ .to(result);
+ }
+ };
+ }
+
+ @Test
+ public void testReceiveTwoPipelinedMessagesFromValidationRead() throws
Exception {
+ assertPipelinedMessagesReceived(1, 2);
+ }
+
+ @Test
+ public void testReceiveThreePipelinedMessages() throws Exception {
+ assertPipelinedMessagesReceived(1, 2, 3);
+ }
+
+ @Test
+ public void testReceivePipelinedMessagesAfterAcknowledgedMessage() throws
Exception {
+ mllpClient.connect();
+ String firstMessage = Hl7TestMessageGenerator.generateMessage(1);
+ result.expectedMessageCount(3);
+ result.message(0).body().isEqualTo(firstMessage);
+ mllpClient.sendFramedData(firstMessage);
+ assertAcknowledgement(1);
+
+ String secondMessage = Hl7TestMessageGenerator.generateMessage(2);
+ String thirdMessage = Hl7TestMessageGenerator.generateMessage(3);
+ result.message(1).body().isEqualTo(secondMessage);
+ result.message(2).body().isEqualTo(thirdMessage);
+ mllpClient.sendFramedDataPipelined(secondMessage, thirdMessage);
+ assertAcknowledgement(2);
+ assertAcknowledgement(3);
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testReceivePipelinedMessageAfterJunkContainingStartOfBlock()
throws Exception {
+ String firstMessage = Hl7TestMessageGenerator.generateMessage(1);
+ String secondMessage = Hl7TestMessageGenerator.generateMessage(2);
+ result.expectedBodiesReceived(firstMessage, secondMessage);
+
+ mllpClient.sendFramedDataPipelined(
+ new byte[] { 'j', 0x0b, 'u', 'n', 'k', 0x1c, 'x' },
firstMessage, secondMessage);
+ assertAcknowledgement(1);
+ assertAcknowledgement(2);
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testReceivePipelinedMessageSpanningReadsAfterLineFeed() throws
Exception {
+ String firstMessage = Hl7TestMessageGenerator.generateMessage(1);
+ StringBuilder secondMessage = new
StringBuilder(Hl7TestMessageGenerator.generateMessage(2));
+ // Larger than the read buffer, so the second message is incomplete
when the first one is processed
+ for (int i = 1; secondMessage.length() < 32 * 1024; i++) {
+
secondMessage.append("OBX|").append(i).append("|TX|NOTE^Note||Lorem ipsum dolor
sit amet||||||F\r");
+ }
+ result.expectedBodiesReceived(firstMessage, secondMessage.toString());
+
+ // Some senders terminate frames with <FS><CR><LF>
+ mllpClient.sendFramedDataPipelined(new byte[] { '\n' }, firstMessage,
secondMessage.toString());
+ assertAcknowledgement(1);
+ assertAcknowledgement(2);
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ }
+
+ private void assertPipelinedMessagesReceived(int... messageNumbers) throws
Exception {
+ String[] messages = new String[messageNumbers.length];
+ result.expectedMessageCount(messageNumbers.length);
+ for (int i = 0; i < messageNumbers.length; i++) {
+ messages[i] =
Hl7TestMessageGenerator.generateMessage(messageNumbers[i]);
+ result.message(i).body().isEqualTo(messages[i]);
+ }
+
+ mllpClient.sendFramedDataPipelined(messages);
+ for (int messageNumber : messageNumbers) {
+ assertAcknowledgement(messageNumber);
+ }
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ }
+
+ private void assertAcknowledgement(int messageNumber) throws Exception {
+ assertThat(mllpClient.receiveFramedData(),
+ containsString(String.format("MSA|AA|%05d", messageNumber)));
+ }
+}
diff --git
a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
index 9ba56232fca8..adc342d28aa7 100644
---
a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
+++
b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
@@ -17,6 +17,7 @@
package org.apache.camel.test.junit.rule.mllp;
import java.io.BufferedOutputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -259,6 +260,59 @@ public class MllpClientResource implements
BeforeEachCallback, AfterEachCallback
}
}
+ /**
+ * Send multiple complete MLLP envelopes in a single socket write without
waiting for their acknowledgements.
+ *
+ * @param hl7Messages messages to send
+ */
+ public void sendFramedDataPipelined(String... hl7Messages) {
+ sendFramedDataPipelined(null, hl7Messages);
+ }
+
+ /**
+ * Send multiple complete MLLP envelopes in a single socket write,
inserting raw data after the first envelope.
+ *
+ * @param dataAfterFirstMessage raw data inserted after the first envelope
+ * @param hl7Messages messages to send
+ */
+ public void sendFramedDataPipelined(byte[] dataAfterFirstMessage,
String... hl7Messages) {
+ if (null == clientSocket) {
+ this.connect();
+ }
+
+ if (!clientSocket.isConnected()) {
+ throw new MllpJUnitResourceException("Cannot send message - client
is not connected");
+ }
+ if (null == outputStream) {
+ throw new MllpJUnitResourceException("Cannot send message - output
stream is null");
+ }
+
+ try {
+ ByteArrayOutputStream framedData = new ByteArrayOutputStream();
+ for (int i = 0; i < hl7Messages.length; i++) {
+ String hl7Message = hl7Messages[i];
+ if (sendStartOfBlock) {
+ framedData.write(START_OF_BLOCK);
+ }
+ framedData.write(hl7Message.getBytes());
+ if (sendEndOfBlock) {
+ framedData.write(END_OF_BLOCK);
+ }
+ if (sendEndOfData) {
+ framedData.write(END_OF_DATA);
+ }
+ if (i == 0 && dataAfterFirstMessage != null) {
+ framedData.write(dataAfterFirstMessage);
+ }
+ }
+ outputStream.write(framedData.toByteArray());
+ outputStream.flush();
+ } catch (IOException e) {
+ log.error("Unable to send pipelined HL7 messages", e);
+ throw new MllpJUnitResourceException("Unable to send pipelined HL7
messages", e);
+ }
+ }
+
public void sendFramedDataInMultiplePackets(String hl7Message, byte
flushByte) {
sendFramedDataInMultiplePackets(hl7Message, flushByte, false);
}
