Fix a possible IllegalReferenceCountException in WebSocket handshake (

#5495)

Motivation:

The following `IllegalReferenceCountException` could be raised during a
HTTP/1 Websocket handshake.
```java
01:54:00.031 [armeria-common-worker-kqueue-2-2] WARN  c.l.a.i.client.PendingExceptionUtil - [id: 0xd2fa853c, L:/10.31.29.254:65140 - R:vks-api-prod.linecorp-dev.com/10.120.97.166:443] Unexpected suppressed exception:
com.linecorp.armeria.common.ClosedSessionException: null
Caused by: io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
	at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:83)
	at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:148)
	at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:101)
	at io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:90)
	at com.linecorp.armeria.client.WebSocketHttp1ClientChannelHandler.channelRead(WebSocketHttp1ClientChannelHandler.java:246)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
	at io.netty.handler.codec.ByteToMessageDecoder.handlerRemoved(ByteToMessageDecoder.java:266)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.remove0(CombinedChannelDuplexHandler.java:608)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.remove(CombinedChannelDuplexHandler.java:593)
	at io.netty.channel.CombinedChannelDuplexHandler.handlerRemoved(CombinedChannelDuplexHandler.java:181)
	at io.netty.channel.AbstractChannelHandlerContext.callHandlerRemoved(AbstractChannelHandlerContext.java:1138)
	at io.netty.channel.DefaultChannelPipeline.callHandlerRemoved0(DefaultChannelPipeline.java:637)
	at io.netty.channel.DefaultChannelPipeline.remove(DefaultChannelPipeline.java:477)
	at io.netty.channel.DefaultChannelPipeline.remove(DefaultChannelPipeline.java:429)
	at com.linecorp.armeria.client.WebSocketHttp1ClientChannelHandler.channelRead(WebSocketHttp1ClientChannelHandler.java:209)
```

- When 101 Switching Protocols response is received,
`WebSocketHttp1ClientChannelHandler` immediately removes
`HttpClientCodec` from the channel pipeline.
https://github.com/line/armeria/blob/75d5af1be6770e5e42690e1b6181c40fed1bf568/core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java#L195
- The removal may emit remaining buffers via `channelRead()`.
https://github.com/netty/netty/blob/06a7e12df9118e3ffbeae3cdc50e638a999d93ec/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java#L266-L267
- WebSocket packets after the HTTP/1 WebSocket handshake are passed to
`WebSocketHttp1ClientChannelHandler`, but the state is still
`NEEDS_HANDSHAKE_RESPONSE`. As a result, the state machine becomes
broken and the `ByteBuf` could be double released.
https://github.com/line/armeria/blob/75d5af1be6770e5e42690e1b6181c40fed1bf568/core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java#L165-L167

Modifications:

- Retain a message before passing the next `ChannelHandler` because it
is always released.
- Remove `HttpClientCodec` after `EMPTY_LAST_CONTENT` is received and
`state` becomes `UPGRADE_COMPLTE` to handle pending messages in the
buffer correctly.

Result:

`IllegalReferenceCountException` no longer occurs during WebSocket
handshake.

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

@@ -164,7 +164,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             switch (state) {
                 case NEEDS_HANDSHAKE_RESPONSE:
                     if (!(msg instanceof HttpObject)) {
-                        ctx.fireChannelRead(msg);
+                        ctx.fireChannelRead(ReferenceCountUtil.retain(msg));
                         return;
                     }
                     if (!(msg instanceof HttpResponse)) {
@@ -191,8 +191,6 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                     res.startResponse();
                     final ResponseHeaders responseHeaders = ArmeriaHttpUtil.toArmeria(nettyRes);
                     if (responseHeaders.status() == HttpStatus.SWITCHING_PROTOCOLS) {
-                        final ChannelPipeline pipeline = ctx.pipeline();
-                        pipeline.remove(HttpClientCodec.class);
                         state = State.NEEDS_HANDSHAKE_RESPONSE_END;
                     }
                     if (!res.tryWriteResponseHeaders(responseHeaders)) {
@@ -205,7 +203,11 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                         failWithUnexpectedMessageType(ctx, msg, EMPTY_LAST_CONTENT.getClass());
                         return;
                     }
+                    // The state should be set to UPGRADE_COMPLETE before removing HttpClientCodec.
+                    // Because pipeline.remove() could trigger channelRead() recursively.
                     state = State.UPGRADE_COMPLETE;
+                    final ChannelPipeline pipeline = ctx.pipeline();
+                    pipeline.remove(HttpClientCodec.class);
                     break;
                 case UPGRADE_COMPLETE:
                     assert msg instanceof ByteBuf;