How does the Camel Netty TCP socket consumer decide how to split incoming data into messages (and is it configurable)?

Question

I'm working with a Camel flow that uses a Netty TCP socket consumer to receive messages from a client program (which is outside of my control). The client should be opening a socket, sending us one message, then closing the socket, but we've been seeing cases where instead of one message Camel is "splitting" the text stream into two parts and trying to process them separately.

So I'm trying to figure out, since you can re-use the same socket for multiple Camel messages, but TCP sockets don't have a built-in concept of "frames" or a standard for message delimiters, how does Camel decide that a complete message has been received and is ready to process? I haven't been able to find a documented answer to this in the Netty component docs (https://camel.apache.org/components/3.15.x/netty-component.html), although maybe I'm missing something.

From playing around with a test script, it seems like one answer is "Camel assumes a message is complete and should be processed if it goes more than 1ms without receiving any input on the socket". Is this a correct statement, and if so is this behavior documented anywhere? Is there any way to change or configure this behavior? Really what I would prefer is for Camel to wait for an ETX character (or a much longer timeout) before processing a message, is that possible to set up?

Here's my test setup:

Camel flow:

from("netty:tcp://localhost:3003")
        .log("Received: ${body}");

Python snippet:

DELAY_MS = 3
def send_msg(sock, msg):
    print("Sending message: <{}>".format(msg))
    if not sock.sendall(msg.encode()) is None:
        print("Message failed to send")
    time.sleep(DELAY_MS / 1000.0)

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
    print("Using DELAY_MS: {}".format(str(DELAY_MS)))
    s.connect((args.hostname, args.port))
    cutoff = int(math.floor(len(args.msg) / 2))
    msg1 = args.msg[:cutoff]
    send_msg(s, msg1)
    msg2 = args.msg[cutoff:]
    send_msg(s, msg2)
    response = s.recv(1024)
except Exception as e:
    print(e)
finally:
    s.close()

I can see that with DELAY_MS=1 Camel logs one single message:

2022-02-21 16:54:40.689  INFO 19429 --- [erExecutorGroup] route1                                   : Received: a long string sent over the socket

But with DELAY_MS=2 it logs two separate messages:

2022-02-21 16:56:12.899  INFO 19429 --- [erExecutorGroup] route1                                   : Received: a long string sen
2022-02-21 16:56:12.899  INFO 19429 --- [erExecutorGroup] route1                                   : Received: t over the socket

Answer 1

After doing some more research, it seems like what I need to do is add a delimiter-based FrameDecoder to the decoders list.

Setting it up like this:

    from("netty:tcp://localhost:3003?sync=true"
         + "&decoders=#frameDecoder,#stringDecoder"
         + "&encoders=#stringEncoder")

where frameDecoder is provided by

@Bean
ChannelHandlerFactory frameDecoder() {
  ByteBuf[] ETX_DELIM = new ByteBuf[] { Unpooled.wrappedBuffer(new byte[] { (char)3 }) };
  return ChannelHandlerFactories.newDelimiterBasedFrameDecoder(1024, ETX_DELIM,
                                                               false, "tcp");
}

seems to do the trick.

On the flip side though, it seems like this will hang indefinitely (or until lower-level TCP timeouts kick in?) if an ETX frame is not received, and I can't figure out any way to set a timeout on the decoder, so would still be eager for input if anyone knows how to do that.

I think the default "timeout" behavior I was seeing might've just been an artifact of Netty's read loop speed -- How does netty determine when a read is complete?