TCP-сервер Netty получает от клиента только 536 байт. Остальное обрезано ⇐ JAVA

[Anonymous]

Я написал приложение TCP-сервера, используя проект Reactor netty. Это простое приложение, которое получает сообщение с запросом массива байтов от клиента, обрабатывает его, а затем возвращает ответное сообщение массива байтов клиенту.
Я столкнулся с проблемой, независимо от того, как много данных, которые на самом деле отправляет клиент, мой TCP-сервер читает не более 536 байт данных. Остальное сокращено. Первые 4 байта сообщения byte[] клиента указывают, какой объем данных был отправлен. Эта длина всегда больше фактических данных, которые я получаю от netty в своем обработчике. Здесь все становится интереснее.
Изначально мне удалось воспроизвести проблему, с которой столкнулся мой клиент, подключившись и отправив данные в приложение TCP Server с моего локального компьютера с помощью написанного мной тестового клиента. . Затем я исправил проблему, добавив дочерние параметры SO_RCVBUF/SO_SNDBUF на свой TCP-сервер и настроив их на больший размер (4096 байт). Теперь я вижу, что полное сообщение принимается и обрабатывается моим TCP-сервером. Однако сообщение моего клиента по-прежнему обрезается даже после добавления этих параметров.

Код: Выделить всё

    TcpServer tcpServer = TcpServer.create();
Optional.of(someTcpServerConfigObject)
.filter(config -> config.isLoopResourcesEnabled())
.ifPresent(enabled -> {
LoopResources loopResources = LoopResources.create ("prefix", 1, 4,true);
tcpServer.runOn(loopResources);
});
DisposableServer someTcpServer = tcpServer
.host("12.123.456.789")
.wiretap(true)
.doOnBind(server -> log.info("Starting listener..."))
.doOnBound(server -> log.info("Listener started on host: {}, port: {}", server.host(), server.port()))
.port(12345)
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.option(ChannelOption.AUTO_CLOSE, false)
.childOption(ChannelOption.TCP_NODELAY,true)
.childOption(ChannelOption.AUTO_CLOSE,false)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childOption(ChannelOption.SO_RCVBUF, 4096)
.childOption(ChannelOption.SO_SNDBUF, 4096)
.doOnConnection(connection -> {
InetSocketAddress socketAddress = (InetSocketAddress) connection.channel().remoteAddress();
log.info("Client has connected. Host: {}, Port: {}",
socketAddress.getAddress().getHostAddress(), socketAddress.getPort());
})
.doOnChannelInit((observer, channel, remoteAddress) ->
channel.pipeline()
.addFirst(new LoggingHandler(MyTcpServer.class))
.addFirst(new TcpServerHandler())
)
.handle((inbound, outbound) ->
inbound
.receive()
.asByteArray()
.flatMap(req -> processRequest(req))
//above processRequest() returns a java.nio.ByteBuffer
//doing rsp.array() to convert to byte[]
.flatMap(rsp -> outbound.sendByteArray(Flux.just(rsp.array()))
.doOnError(throwable -> log.error("Error processing the request: {}", throwable.getMessage(),throwable))
).bindNow();
someTcpServer.onDispose().block();
}

А затем ниже находится класс TcpServerHandler, который я добавил в качестве обработчика выше.

Код: Выделить всё

@Slf4j
public class TcpServerHandler extends ChannelDuplexHandler {

private final AtomicLong startTime = new AtomicLong(0L);
private final AtomicLong endTime = new AtomicLong(0L);

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
byte[] data = HexFormat.of().formatHex(ByteBufUtil.getBytes((ByteBuf) msg));
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
log.info("Receiving message from: Host: {}, Port: {}.  Data: {}", socketAddress.getAddress().getHostAddress(),
socketAddress.getPort(), data);
byte[] byteArrayContainingFourByteLength = new byte[4];
System.arraycopy(data, 0, byteArrayContainingFourByteLength, 0, 4);
ByteBuffer wrapped = ByteBuffer.wrap(byteArrayContainingFourByteLength);
short actualLength = wrapped.getShort();
log.info("# of bytes received from netty: {}", data.length);
log.info("# of bytes client actually sent: {}", actualLength);
startTime.set(System.nanoTime());
ctx.fireChannelRead(msg);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
endTime.set(System.nanoTime());
log.info("Took {} ms to process", Duration.ofNanos(endTime.get() - startTime.get()).toMillis()))
super.write(ctx, msg, promise);
}
}

Пожалуйста, дайте мне знать, если есть какие-либо другие параметры/дочерние параметры/обработчики/и т. д. на моем TCP-сервере я могу настроить решение этой проблемы. Проведя небольшое исследование, я нашел страницу в Википедии, посвященную максимальному размеру сегмента. Там также упоминается тот же предел в 536 байт, поэтому мне интересно, связана ли проблема моего клиента и могу ли я каким-то образом настроить это через netty.

Подробнее здесь: https://stackoverflow.com/questions/788 ... -truncated