NPE in websocket server if client has keepAlive

[tsachev]

If I create my websocket client with keepAlive like this

RSocketFactory.connect().keepAlive().transport(WebsocketClientTransport.create(7878)).start();

the server throws npe (see below) when it receives the keep alive frame.
The strange thing is that I do not see this with tcp transport at least for ping pong tests.

I'm just guessing that the problem is somewhere in WebsocketDuplexConnection or probably in FragmentationDuplexConnection if the websocket frames are not sized same as tcp?

java.lang.NullPointerException
	at io.rsocket.frame.FrameHeaderFlyweight.frameType(FrameHeaderFlyweight.java:191)
	at io.rsocket.Frame.getType(Frame.java:219)
	at io.rsocket.Frame.ensureFrameType(Frame.java:595)
	at io.rsocket.Frame$Keepalive.hasRespondFlag(Frame.java:586)
	at io.rsocket.RSocketServer.lambda$handleKeepAliveFrame$17(RSocketServer.java:394)
	at reactor.core.publisher.MonoRunnable.subscribe(MonoRunnable.java:40)
	at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
	at reactor.core.publisher.Mono.subscribe(Mono.java:3008)
	at reactor.core.publisher.FluxMergeSequential$MergeSequentialMain.onNext(FluxMergeSequential.java:230)
	at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:238)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:555)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:631)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.onNext(FluxGroupBy.java:671)
	at reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:205)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108)
	at reactor.ipc.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:211)
	at reactor.ipc.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:326)
	at reactor.ipc.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:319)
	at reactor.ipc.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:381)
	at reactor.ipc.netty.http.server.HttpServerWSOperations.onInboundNext(HttpServerWSOperations.java:107)
	at reactor.ipc.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:132)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:141)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
	at java.lang.Thread.run(Thread.java:748)
java.lang.NullPointerException
	at io.rsocket.frame.FrameHeaderFlyweight.frameType(FrameHeaderFlyweight.java:191)
	at io.rsocket.Frame.getType(Frame.java:219)
	at io.rsocket.Frame.ensureFrameType(Frame.java:595)
	at io.rsocket.Frame$Keepalive.hasRespondFlag(Frame.java:586)
	at io.rsocket.RSocketServer.lambda$handleKeepAliveFrame$17(RSocketServer.java:394)
	at reactor.core.publisher.MonoRunnable.subscribe(MonoRunnable.java:40)
	at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
	at reactor.core.publisher.Mono.subscribe(Mono.java:3008)
	at reactor.core.publisher.FluxMergeSequential$MergeSequentialMain.onNext(FluxMergeSequential.java:230)
	at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:238)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:555)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:631)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.onNext(FluxGroupBy.java:671)
	at reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:205)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108)
	at reactor.ipc.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:211)
	at reactor.ipc.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:326)
	at reactor.ipc.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:319)
	at reactor.ipc.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:381)
	at reactor.ipc.netty.http.server.HttpServerWSOperations.onInboundNext(HttpServerWSOperations.java:107)
	at reactor.ipc.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:132)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:141)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
	at java.lang.Thread.run(Thread.java:748)
java.lang.NullPointerException
	at io.rsocket.frame.FrameHeaderFlyweight.frameType(FrameHeaderFlyweight.java:191)
	at io.rsocket.Frame.getType(Frame.java:219)
	at io.rsocket.Frame.ensureFrameType(Frame.java:595)
	at io.rsocket.Frame$Keepalive.hasRespondFlag(Frame.java:586)
	at io.rsocket.RSocketServer.lambda$handleKeepAliveFrame$17(RSocketServer.java:394)
	at reactor.core.publisher.MonoRunnable.subscribe(MonoRunnable.java:40)
	at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
	at reactor.core.publisher.Mono.subscribe(Mono.java:3008)
	at reactor.core.publisher.FluxMergeSequential$MergeSequentialMain.onNext(FluxMergeSequential.java:230)
	at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:238)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:555)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:631)
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.onNext(FluxGroupBy.java:671)
	at reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:205)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108)
	at reactor.ipc.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:211)
	at reactor.ipc.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:326)
	at reactor.ipc.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:319)
	at reactor.ipc.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:381)
	at reactor.ipc.netty.http.server.HttpServerWSOperations.onInboundNext(HttpServerWSOperations.java:107)
	at reactor.ipc.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:132)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:297)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:413)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:265)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:141)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
	at java.lang.Thread.run(Thread.java:748)

[mostroverkhov]

@tsachev Likely problem is in Core - Frame is already released when Keep-Alive handling code tries to decode It (one, two, three). This can happen when Mono's lambda inside handleKeepAliveFrame is executed on non-synchronous scheduler.

[tsachev]

The release in the finally block is called before anyone has a chance to subscribe to the returned mono. So even on synchronous scheduler it won't work if the ref count of frame's content was 1. With tcp it's 2 but I don't know exactly why.

[mostroverkhov]

@tsachev 1 refCount is by reactor-netty, which pools Bytebuffs, another one is by RSocket-java. Also, I think this issue was fixed in #467, can you confirm?

[yschimke]

Closing as I'm using this in rsocket-cli