reactor/reactor-netty

Reactive Network Driver based on Netty and Reactor

Sourabh

@dragneelfps

Hi all, I was trying to implement mysql tcp protocol with netty. My Enveloper is not getting executed if I do channel.addLast().

val conn = TcpClient.newConnection()
        .host("localhost")
        .port(3306)
        .handle { inbound, outbound -> inbound.receive().then() }
        .doOnChannelInit { a, channel, _ ->
            channel.pipeline()
                .addFirst(LoggingHandler(LogLevel.TRACE))
                .addLast(EnvelopeHandler())

        }
        .connectNow()

    conn.onDispose()
        .block()

Please help

2 replies

pipeline:

DefaultChannelPipeline{(LoggingHandler#0 = io.netty.handler.logging.LoggingHandler), (TransportConfig$TransportChannelInitializer#0 = reactor.netty.transport.TransportConfig$TransportChannelInitializer), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler), (EnvelopeHandler#0 = EnvelopeHandler)}

Sourabh

@dragneelfps

Does LengthFieldBasedFrameDecoder buffers until UnadjustedFrameLength amount of packets are received, before passing to next handler in the pipeline? If not, how do I make sure that next handler have the full message?

Mikael Elm

@mickeelm

At my client we have a use case where we are to implement a legacy protocol. Our stack in general is Spring Boot/microservice-based, and this is a special case external implementation that needs to be integrated with Spring (Boot).

With the protocol (text based, something like 043G00005ABC but potentially way longer messages) we have to be able to keep track of/route messages based on ids/characteristics. It's not request-reply but rather request-reply-request-reply, potentially many times and over several seconds. It's also possible that different messages are sent in between/in parallel. So with the commands A and B, the following is possible:

A1 ->
<- A2
A3 ->
<- A4
A5 ->
B1 ->
<- A6
<- B2
A7 ->
<- A8
B3 ->
<- B4
A9 ->

and so on. I have used some Reactor but not dug into Reactor Netty anything really. Does it seem like a good use case for Reactor Netty, especially given the Spring Boot-integration? Other frameworks that are Spring friendly and deals with these kind of protocols? Any tips here are highly appreciated!

JonahCui

@yoster0520

any one know this?

```

[78cb6ed3] Error [java.lang.UnsupportedOperationException] for HTTP GET "/ladder/api/user/getMediaRecommend?page_size=50&t_sign=0", but ServerHttpResponse already committed (200 OK)

1 reply

any one know this????

[78cb6ed3] Error [java.lang.UnsupportedOperationException] for HTTP GET "/ladder/api/user/getMediaRecommend?page_size=50&t_sign=0", but ServerHttpResponse already committed (200 OK)

I use spring cloud gateway, the version is :

reactor-netty: 0.9.15.RELEASE
spring-cloud-gateway: 2.2.8.RELEASE
spribg-boot: 2.2.12.RELEASE

dileep

@dileepkmandapam_twitter

Hello Team, How to receive channel information from NettyInbound while handling the requests.

This message was deleted

                .handle(new BiFunction<NettyInbound, NettyOutbound, Publisher<Void>>() {
                    @Override
                    public Publisher<Void> apply(NettyInbound nettyInbound, NettyOutbound nettyOutbound) {
                        Flux<ByteBuf> byteBufMono = nettyInbound.receive().retain().asString().flatMap(handler::handleCommands).metrics();
                        return nettyOutbound.send(byteBufMono, byteBuf -> true);
                    }
                });

I am planning to collect metrics per service type.

@violetagg : Any poniters

Violeta Georgieva

@violetagg

@dileepkmandapam_twitter What do you need from the channel as information?

basically you have life cycle callbacks where you can obtain the connection info

https://projectreactor.io/docs/netty/release/reference/index.html#_lifecycle_callbacks_2

or you can use withConnection callback https://projectreactor.io/docs/netty/release/api/reactor/netty/NettyInbound.html#withConnection-java.util.function.Consumer-

dileep

@dileepkmandapam_twitter

@violetagg In the above handler, How do we fetch channel ip address and certificate.

Violeta Georgieva

@violetagg

@dileepkmandapam_twitter Can you try this nettyInbound.withConnection(conn -> conn.channel()…)

dileep

@dileepkmandapam_twitter

@violetagg I think I need to use reactor context to pass information.

1 reply

dileep

@dileepkmandapam_twitter

@violetagg : Just one more question, Do we need to explicitly release the ByteBuf here.

.handle(new BiFunction<NettyInbound, NettyOutbound, Publisher<Void>>() {
                    @Override
                    public Publisher<Void> apply(NettyInbound nettyInbound, NettyOutbound nettyOutbound) {
                        Flux<ByteBuf> byteBufMono = nettyInbound.receive().retain().asString().flatMap(handler::handleCommands).metrics();
                        return nettyOutbound.send(byteBufMono, byteBuf -> true);
                    }
                });

4 replies

dileep

@dileepkmandapam_twitter

@violetagg : Typically handler is implemented in the following manner.

    private ByteBuf convert(Object resp) {
        ByteBuf message = null;
        try {
            var responsePayload = mapper.writeValueAsString(resp);

            message = Unpooled.buffer(responsePayload.length() + 8);
            message.writeInt(1);//Tag
            var bytes = responsePayload.getBytes(StandardCharsets.UTF_8);
            message.writeInt(bytes.length);//length
            message.writeBytes(bytes);
        } catch (JsonProcessingException e) {
            log.error("Failed to serialize the resp", e);
        }
        return message;
    }

    private Mono<ByteBuf> handleGet(LookUpRequest request) {
        log.info("Handling:{}", request);
        return reactiveStoreOperations.get(request).name("get").metrics().map(resp -> {
            var lookupResp = new LookUpResponse();
            lookupResp.setValue(resp);
            log.info("GET::Resp:{}", resp);
            return convert(lookupResp);
        }).switchIfEmpty(Mono.fromSupplier(() -> {
            log.error("Received empty response for req:{}", request);
            return (Unpooled.buffer(1));
        }));

@violetagg Here is the handlerCommand implementation.

    Mono<ByteBuf> handleCommands(String json) {

        log.trace("Received payload:{}", json);
        RequestObject requestObject = null;
        try {
            requestObject = mapper.readerFor(RequestObject.class).readValue(json);
        } catch (JsonProcessingException e) {
            log.error("Failed to de-serialize json:" + json, e);
            return Mono.empty();
        }
        if (wrapperObject.getGetLookUpRequest() != null) {
            return handleGet(wrapperObject.getGetLookUpRequest());
        } 
        log.error("Unknown request:{}", wrapperObject);
        return Mono.empty();
    }

dileep

@dileepkmandapam_twitter

Sometimes I am seeing below ERRORs in the logs.

1:20:02.786 [reactor-tcp-nio-7] ERROR [{}] io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
    io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:385)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)
    io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:115)
    io.netty.handler.ssl.SslHandler.allocate(SslHandler.java:2223)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1383)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:829)

Violeta Georgieva

@violetagg

@dileepkmandapam_twitter Please run with -Dio.netty.leakDetection.level=paranoid and increase the logging level for _reactor.netty.channel.LeakDetection to debug

dileep

@dileepkmandapam_twitter

@violetagg I got bunch of leak detector logs. Here is the sample.

14:24:58.661 [reactor-tcp-nio-8] ERROR [{}] io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
    io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:88)
    reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:284)
    reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
    reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:401)
    reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)

2 replies

dileep

@dileepkmandapam_twitter

14:24:58.661 [reactor-tcp-nio-8] ERROR [{}] io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
    io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:88)
    reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:284)
    reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
    reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:401)
    reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThrea

#2:
    io.netty.buffer.AdvancedLeakAwareByteBuf.readCharSequence(AdvancedLeakAwareByteBuf.java:526)
    reactor.netty.ByteBufFlux.lambda$asString$6(ByteBufFlux.java:262)
    reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:102)
    reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
    reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:220)
    reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:280)
    reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
    reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:401)
    reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutor

#3:
    reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:840)
    reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:220)
    reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:280)
    reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
    reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:401)
    reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:829)

#4:
    Hint: [id:abd7e848, L:/127.0.0.1:7194 - R:/127.0.0.1:64527] Receiver reactor.core.publisher.FluxMap$MapConditionalSubscriber will handle the message from this point
    reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:272)
    reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
    reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:401)
    reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:829)

#5:
    Hint: [id:abd7e848, L:/127.0.0.1:7194 - R:/127.0.0.1:64527] Buffered ByteBuf in the inbound buffer queue
    reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:382)
    reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:401)
    reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:829)

#6:
    Hint: 'reactor.right.reactiveBridge' will handle the message from this point.
    io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:116)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:829)

Created at:
    io.netty.buffer.SimpleLeakAwareByteBuf.unwrappedDerived(SimpleLeakAwareByteBuf.java:143)
    io.netty.buffer.SimpleLeakAwareByteBuf.retainedSlice(SimpleLeakAwareByteBuf.java:57)
    io.netty.buffer.AdvancedLeakAwareByteBuf.retainedSlice(AdvancedLeakAwareByteBuf.java:95)
    io.netty.handler.codec.LengthFieldBasedFrameDecoder.extractFrame(LengthFieldBasedFrameDecoder.java:498)
    io.netty.handler.codec.LengthFieldBasedFrameDecoder.decode(LengthFieldBasedFrameDecoder.java:437)
    io.netty.handler.codec.LengthFieldBasedFrameDecoder.decode(LengthFieldBasedFrameDecoder.java:332)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:271)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    reactor.netty.channel.ChannelMetricsHandler.channelRead(ChannelMetricsHandler.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1534)
    io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1283)
    io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1330)
    io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:829)

Murdock

@phnxfire

hi All!

Very basic question - I have created a netty reactor websocket and it's working fine... last step I need to make it able to receive secure connections using -> wss:// - how do I do it?

4 replies

Murdock

@phnxfire

I have a client developed in Swift which is sending ping messages to my reactive websocket server.
The swift client is sending ping messages like this:

@objc private func sendSocketPings(_ timer: Timer) {
if let stackSocket = self.stackSocket {
Log.debug("pinging stackSocket")
stackSocket.write(ping: Data())
}
}

However the back end doesn't get any message, not a single one on the java back end which is of type PING.

public Mono<Void> handle(WebSocketSession webSocketSession) {
webSocketSession
.receive()
.map(webSocketMessage ->{
log.info(" Stack Handler - Web socket receive message");
log.info(" Stack Handler - WebSocketMessage Type : {}", webSocketMessage.getType());

        if (webSocketMessage.getType()==WebSocketMessage.Type.PING){

30 replies

Tomasz Adamski

@tmszdmsk

Hi,

I have a problem after upgrading reactor from 0.9.20 -> 1.0.7 (together with spring boot).

On client connection closed there is AbortedException raised by reactor-netty which is "caused" by this recent change: reactor/reactor-netty@8a1188e

Problem is that, despite supporting "AbortedException" in spring-framework (org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler). It is being bubbled up and produces http response 500.

Looks like the problem is somewhere in the integration between Springs HttpResponse and Reactor Netty.

When client connection drop happens this if (response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR)) { is evaluated before checking if it isDisconnectedClientError and it allows to set the response status which I think it shouldn't. (https://github.com/spring-projects/spring-framework/blob/main/spring-web/src/main/java/org/springframework/web/server/adapter/HttpWebHandlerAdapter.java#L292)

Are you aware of this and is it the right place to ask?

10 replies

ruslan

@unoexperto

Folks, is there way to define error handler in reactor-netty server instead of doing it manually with Mono.onErrorResume() in every route handler ? Basically I want every exception in every route to be wrapped into standard json response.

8 replies

alwaystheobvious

@alwaystheobvious

Hi!

alwaystheobvious

@alwaystheobvious

I use tcpserver and tcpclient and on the client side it seems that I'm unable to create requests and receive the incoming data in the same place. Take for example a class named MyClient where I initialize the client instance and start it , set up a basic data handler inside handle(inbound, outbound) and then there's another class named ImageLoader. I'd like to be able to create a request call from ImageLoader class and then get the request data there. Is this possible? Please help me...been searching for 2 days for a solution. I've built a workaround it using callbacks and queues but i don't think it's the best solution.

69 replies

Kanai

@kanai0618

Hi Team, I am using Spring cloud gateway in our gateway service. Recently our service is serving more traffic, and started seeing connection prematurely closed issues. Then i have set the max-idle-time. once i have set then i saw few Connection has been closed BEFORE send operation. could you please help me ?

35 replies

jcubells

@jcubells

Hi ! I'm using reactor-netty 0.9.12 and running on an issue in production environment.
The requests from my client are received by my server after a delay, causing the client to timeout (if this delay is too important).
From what I can see, the client establishes connection, and sends the request.
Then, in tcpdump on server-side, I can see there is an incoming HTTP request, but the HttpServer in netty doesn't log anything.
After 5s, my client sends a timeout (tcpdump shows the "END" signal being sent)
After a while, my HttpServer wiretap finally logs the incoming query, but then throws an error (since the client timeout closed the socket)

I have no idea about what could be causing the issue, and what to do to get more information about the root cause and/or fix the issue. Any help would be welcome

6 replies

Amandeep Singh

@Man-Has-No-Name

Hi Team,
We are using reactor netty (1.0.7 latest) in our organisation and we have a use case in which we want to write tests to make sure the connection gets aborted on the TCP level by server so that the client can perform the retry.
Is there any server hook or implementation we can use to configure this?

7 replies

alwaystheobvious

@alwaystheobvious

Hi Team, how can I get the connection or channel id with each message from client? (tcp server and tcp client), the code:

                           .handle((in, out) ->

                                    out.sendObject(
                                            in.receiveObject().flatMap(msg -> { 
                                                Connection connection = ?? //get connection / channel id to do some work with it... but how?? 
                                                return Flux.empty();
                                            }))

28 replies

ctlove0523

@ctlove0523

Hi Team,I use HttpClient send request to outside server get some useful info,when http request success，but outside server return error message,like resource not exist（http status code 404），how can i graceful handle this:

@Override
    public Mono<IotInnerInstance> showInnerIotInstance(String instanceId) {
        return client.get()
            .uri(requestUri(instanceId))
            .responseSingle(new BiFunction<HttpClientResponse, ByteBufMono, Mono<IotInnerInstance>>() {
                @Override
                public Mono<IotInnerInstance> apply(HttpClientResponse response, ByteBufMono byteBufMono) {
                    if (HttpClientResponseUtils.isSuccessful(response)) {
                        return byteBufMono.asString(StandardCharsets.UTF_8)
                            .map(new Function<String, IotInnerInstance>() {
                                @Override
                                public IotInnerInstance apply(String s) {
                                    return JacksonUtil.json2Pojo(s, IotInnerInstance.class);
                                }
                            });
                    }
                    // can not convert to IotInnerInstance
                    // byteBufMono represent error response（http status code not 2xx),I want return this error message to my 
                    // client,how can i do this?
                    return Mono.error(new Exception("response from iomgr failed,response code is " + response.status()));

                }

            });
    }

4 replies

Murdock

@phnxfire

hi all, I'm getting this:

EAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.\nRecent access records: \nCreated at:\n\tio.netty.buffer.UnpooledByteBufAllocator.newDirectBuffer(UnpooledByteBufAllocator.java:96)\n\tio.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)\n\tio.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)\n\tio.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:115)\n\torg.springframework.core.io.buffer.NettyDataBufferFactory.allocateBuffer(NettyDataBufferFactory.java:71)\n\torg.springframework.core.io.buffer.NettyDataBufferFactory.allocateBuffer(NettyDataBufferFactory.java:39)\n\tcom.apple.ist.retail.talk.ws.redis.RedisStoreListener.getDataBuffer(RedisStoreListener.java:66)\n\tcom.apple.ist.retail.talk.ws.redis.RedisStoreListener.message(RedisStoreListener.java:31)\n\tio.lettuce.core.pubsub.PubSubEndpoint.notifyListeners(PubSubEndpoint.java:217)\n\tio.lettuce.core.pubsub.PubSubEndpoint.notifyMessage(PubSubEndpoint.java:206)\n\tio.lettuce.core.pubsub.PubSubCommandHandler.doNotifyMessage(PubSubCommandHandler.java:292)\n\tio.lettuce.core.pubsub.PubSubCommandHandler.notifyPushListeners(PubSubCommandHandler.java:223)\n\tio.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:620)\n\tio.lettuce.core.pubsub.PubSubCommandHandler.decode(PubSubCommandHandler.java:112)\n\tio.lettuce.core.protocol.CommandHandler.channelRead(CommandHandler.java:572)\n\tio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\tio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\tio.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\tio.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)\n\tio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\tio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\tio.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)\n\tio.netty.channel.kqueue.AbstractKQueueStreamChannel$KQueueStreamUnsafe.readReady(AbstractKQueueStreamChannel.java:544)\n\tio.netty.channel.kqueue.AbstractKQueueChannel$AbstractKQueueUnsafe.readReady(AbstractKQueueChannel.java:382)\n\tio.netty.channel.kqueue.KQueueEventLoop.processReady(KQueueEventLoop.java:211)\n\tio.netty.channel.kqueue.KQueueEventLoop.run(KQueueEventLoop.java:289)\n\tio.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)\n\tio.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)\n\tio.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\tjava.base/java.lang.Thread.run(Thread.java:834)

Not sure where/when I should call ByteBuf.release

I'm dealing with reactive websockets and DataBuffer

Where communities thrive

People

Repo info

Activity