Croway commented on code in PR #26259:
URL: https://github.com/apache/camel/pull/26259#discussion_r3978274124
##########
components/camel-mllp/src/main/java/org/apache/camel/component/mllp/internal/TcpSocketConsumerRunnable.java:
##########
@@ -217,6 +200,77 @@ public void run() {
}
}
+ /**
+ * 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;
+ for (int i = 0; i < trailingData.length; i++) {
+ if (trailingData[i] == MllpProtocolConstants.START_OF_BLOCK) {
+ if (i == 0 ||
mllpBuffer.isCompleteEnvelopeAt(trailingDataOffset + i)) {
Review Comment:
**Pipelined message still lost when an `<LF>` follows `<FS><CR>`**
When out-of-band bytes come before the next `<VT>` and that message isn't
fully buffered yet, this condition is false, so the `<VT>` is skipped and the
partial message is discarded. The rest of the message then arrives with no
`<VT>` and `readSocketInputStream` drops it too, so it is never routed or ACKed.
This can happen in practice with senders that end frames with `<FS><CR><LF>`
(for example, files queued with a trailing newline, see
nextgenhealthcare/connect#510) when they pipeline a message larger than what is
left of the read (default `minBufferSize=2048`). The server logs `Ignoring
trailing out-of-band data`, followed by two `ignoring N bytes received before
START_OF_BLOCK` warnings.
Suggestion: skip leading `CR`/`LF` bytes before applying the `i == 0` rule.
The partial message is then kept, and junk that contains a `<VT>` keeps the
current behaviour.
<details><summary>Reproducer: <code>lfTrailerSecondMessageSpansReads</code>
fails on this PR's head (SocketTimeoutException waiting for ACK 2), the control
passes</summary>
```java
package org.apache.camel.component.mllp;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
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.junit6.CamelTestSupport;
import org.apache.camel.test.mllp.Hl7TestMessageGenerator;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class MllpPipelinedTrailingLineFeedTest extends CamelTestSupport {
final int port = AvailablePortFinder.getNextAvailable();
@EndpointInject("mock://result")
MockEndpoint result;
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
fromF("mllp://localhost:%d", port).to(result);
}
};
}
// Fails: message 2 is logged as "Ignoring trailing out-of-band data"
and never ACKed
@Test
public void lfTrailerSecondMessageSpansReads() throws Exception {
runPipelined(new byte[] { 0x1c, 0x0d, 0x0a });
}
// Passes: same traffic without the LF
@Test
public void noLfTrailerSecondMessageSpansReads() throws Exception {
runPipelined(new byte[] { 0x1c, 0x0d });
}
private void runPipelined(byte[] trailer) throws Exception {
String first = Hl7TestMessageGenerator.generateMessage(1);
StringBuilder second = new
StringBuilder(Hl7TestMessageGenerator.generateMessage(2));
for (int i = 1; second.length() < 16 * 1024; i++) {
// larger than what is left of the default 2048-byte first read
second.append("OBX|").append(i).append("|TX|NOTE^Note||Lorem
ipsum dolor sit amet||||||F\r");
}
result.expectedBodiesReceived(first, second.toString());
ByteArrayOutputStream wire = new ByteArrayOutputStream();
for (String hl7 : new String[] { first, second.toString() }) {
wire.write(0x0b);
wire.writeBytes(hl7.getBytes(StandardCharsets.ISO_8859_1));
wire.writeBytes(trailer);
}
try (Socket socket = new Socket("localhost", port)) {
socket.setSoTimeout(10000);
OutputStream out = socket.getOutputStream();
out.write(wire.toByteArray()); // single write: pipelined
out.flush();
InputStream in = socket.getInputStream();
assertTrue(readFrame(in).contains("MSA|AA|00001"));
assertTrue(readFrame(in).contains("MSA|AA|00002")); //
SocketTimeoutException with the LF trailer
}
MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
}
private static String readFrame(InputStream in) throws Exception {
ByteArrayOutputStream bytes = new ByteArrayOutputStream();
int b;
while ((b = in.read()) != 0x1c) {
if (b == -1) {
throw new IllegalStateException("connection closed");
}
bytes.write(b);
}
in.read(); // <CR>
return bytes.toString(StandardCharsets.ISO_8859_1);
}
}
```
</details>
_Claude Code on behalf of Croway_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
