#9944 Merge WebSocketCloseFrameHandler into WebSocketProtocolHandler … (#9967)

…in order to minimize pipeline

Motivation:

Handling of `WebSocketCloseFrame` is part of websocket protocol, so it's logical to put it within the `WebSocketProtocolHandler`. Also, removal of `WebSocketCloseFrameHandler` will decrease the channel pipeline.

Modification:

- `WebSocketCloseFrameHandler` code merged into `WebSocketProtocolHandler`. `WebSocketCloseFrameHandler` not added to the pipeline anymore
- Added additional constructor to `WebSocketProtocolHandler`
- `WebSocketProtocolHandler` now implements `ChannelOutboundHandler` and implements basic methods from it

Result:

`WebSocketCloseFrameHandler` is no longer used.

Fixes https://github.com/netty/netty/issues/9944

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientProtocolHandler.java

@@ -78,7 +78,8 @@ public class WebSocketClientProtocolHandler extends WebSocketProtocolHandler {
      *            Client protocol configuration.
      */
     public WebSocketClientProtocolHandler(WebSocketClientProtocolConfig clientConfig) {
-        super(checkNotNull(clientConfig, "clientConfig").dropPongFrames());
+        super(checkNotNull(clientConfig, "clientConfig").dropPongFrames(),
+              clientConfig.sendCloseFrame(), clientConfig.forceCloseTimeoutMillis());
         this.handshaker = WebSocketClientHandshakerFactory.newHandshaker(
             clientConfig.webSocketUri(),
             clientConfig.version(),
@@ -378,9 +379,5 @@ public class WebSocketClientProtocolHandler extends WebSocketProtocolHandler {
             ctx.pipeline().addBefore(ctx.name(), Utf8FrameValidator.class.getName(),
                     new Utf8FrameValidator());
         }
-        if (clientConfig.sendCloseFrame() != null) {
-            cp.addBefore(ctx.name(), WebSocketCloseFrameHandler.class.getName(),
-                new WebSocketCloseFrameHandler(clientConfig.sendCloseFrame(), clientConfig.forceCloseTimeoutMillis()));
-        }
     }
 }

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketCloseFrameHandler.java

@@ -1,97 +0,0 @@
-/*
- * Copyright 2019 The Netty Project
- *
- * The Netty Project licenses this file to you under the Apache License,
- * version 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at:
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-package io.netty.handler.codec.http.websocketx;
-
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelOutboundHandlerAdapter;
-import io.netty.channel.ChannelPromise;
-import io.netty.util.ReferenceCountUtil;
-import io.netty.util.concurrent.ScheduledFuture;
-import io.netty.util.internal.ObjectUtil;
-
-import java.nio.channels.ClosedChannelException;
-import java.util.concurrent.TimeUnit;
-
-/**
- * Send {@link CloseWebSocketFrame} message on channel close, if close frame was not sent before.
- */
-final class WebSocketCloseFrameHandler extends ChannelOutboundHandlerAdapter {
-    private final WebSocketCloseStatus closeStatus;
-    private final long forceCloseTimeoutMillis;
-    private ChannelPromise closeSent;
-
-    WebSocketCloseFrameHandler(WebSocketCloseStatus closeStatus, long forceCloseTimeoutMillis) {
-        this.closeStatus = ObjectUtil.checkNotNull(closeStatus, "closeStatus");
-        this.forceCloseTimeoutMillis = forceCloseTimeoutMillis;
-    }
-
-    @Override
-    public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
-        if (!ctx.channel().isActive()) {
-            ctx.close(promise);
-            return;
-        }
-        if (closeSent == null) {
-            write(ctx, new CloseWebSocketFrame(closeStatus), ctx.newPromise());
-        }
-        flush(ctx);
-        applyCloseSentTimeout(ctx);
-        closeSent.addListener(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(ChannelFuture future) {
-                ctx.close(promise);
-            }
-        });
-    }
-
-    @Override
-    public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
-        if (closeSent != null) {
-            ReferenceCountUtil.release(msg);
-            promise.setFailure(new ClosedChannelException());
-            return;
-        }
-        if (msg instanceof CloseWebSocketFrame) {
-            promise = promise.unvoid();
-            closeSent = promise;
-        }
-        super.write(ctx, msg, promise);
-    }
-
-    private void applyCloseSentTimeout(ChannelHandlerContext ctx) {
-        if (closeSent.isDone() || forceCloseTimeoutMillis < 0) {
-            return;
-        }
-
-        final ScheduledFuture<?> timeoutTask = ctx.executor().schedule(new Runnable() {
-            @Override
-            public void run() {
-                if (!closeSent.isDone()) {
-                    closeSent.tryFailure(new WebSocketHandshakeException("send close frame timed out"));
-                }
-            }
-        }, forceCloseTimeoutMillis, TimeUnit.MILLISECONDS);
-
-        closeSent.addListener(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(ChannelFuture future) {
-                timeoutTask.cancel(false);
-            }
-        });
-    }
-}

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandler.java

@@ -16,14 +16,27 @@
 package io.netty.handler.codec.http.websocketx;
 
 
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandler;
+import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.MessageToMessageDecoder;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ScheduledFuture;
 
+import java.net.SocketAddress;
+import java.nio.channels.ClosedChannelException;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
-abstract class WebSocketProtocolHandler extends MessageToMessageDecoder<WebSocketFrame> {
+abstract class WebSocketProtocolHandler extends MessageToMessageDecoder<WebSocketFrame>
+        implements ChannelOutboundHandler {
 
     private final boolean dropPongFrames;
+    private final WebSocketCloseStatus closeStatus;
+    private final long forceCloseTimeoutMillis;
+    private ChannelPromise closeSent;
 
     /**
      * Creates a new {@link WebSocketProtocolHandler} that will <i>drop</i> {@link PongWebSocketFrame}s.
@@ -40,7 +53,15 @@ abstract class WebSocketProtocolHandler extends MessageToMessageDecoder<WebSocke
      *            {@code true} if {@link PongWebSocketFrame}s should be dropped
      */
     WebSocketProtocolHandler(boolean dropPongFrames) {
+        this(dropPongFrames, null, 0L);
+    }
+
+    WebSocketProtocolHandler(boolean dropPongFrames,
+                             WebSocketCloseStatus closeStatus,
+                             long forceCloseTimeoutMillis) {
         this.dropPongFrames = dropPongFrames;
+        this.closeStatus = closeStatus;
+        this.forceCloseTimeoutMillis = forceCloseTimeoutMillis;
     }
 
     @Override
@@ -65,6 +86,94 @@ abstract class WebSocketProtocolHandler extends MessageToMessageDecoder<WebSocke
         }
     }
 
+    @Override
+    public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
+        if (closeStatus == null || !ctx.channel().isActive()) {
+            ctx.close(promise);
+        } else {
+            if (closeSent == null) {
+                write(ctx, new CloseWebSocketFrame(closeStatus), ctx.newPromise());
+            }
+            flush(ctx);
+            applyCloseSentTimeout(ctx);
+            closeSent.addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture future) {
+                    ctx.close(promise);
+                }
+            });
+        }
+    }
+
+    @Override
+    public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        if (closeSent != null) {
+            ReferenceCountUtil.release(msg);
+            promise.setFailure(new ClosedChannelException());
+            return;
+        }
+        if (msg instanceof CloseWebSocketFrame) {
+            promise = promise.unvoid();
+            closeSent = promise;
+        }
+        ctx.write(msg, promise);
+    }
+
+    private void applyCloseSentTimeout(ChannelHandlerContext ctx) {
+        if (closeSent.isDone() || forceCloseTimeoutMillis < 0) {
+            return;
+        }
+
+        final ScheduledFuture<?> timeoutTask = ctx.executor().schedule(new Runnable() {
+            @Override
+            public void run() {
+                if (!closeSent.isDone()) {
+                    closeSent.tryFailure(new WebSocketHandshakeException("send close frame timed out"));
+                }
+            }
+        }, forceCloseTimeoutMillis, TimeUnit.MILLISECONDS);
+
+        closeSent.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) {
+                timeoutTask.cancel(false);
+            }
+        });
+    }
+
+    @Override
+    public void bind(ChannelHandlerContext ctx, SocketAddress localAddress,
+                     ChannelPromise promise) throws Exception {
+        ctx.bind(localAddress, promise);
+    }
+
+    @Override
+    public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress,
+                        SocketAddress localAddress, ChannelPromise promise) throws Exception {
+        ctx.connect(remoteAddress, localAddress, promise);
+    }
+
+    @Override
+    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise)
+            throws Exception {
+        ctx.disconnect(promise);
+    }
+
+    @Override
+    public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+        ctx.deregister(promise);
+    }
+
+    @Override
+    public void read(ChannelHandlerContext ctx) throws Exception {
+        ctx.read();
+    }
+
+    @Override
+    public void flush(ChannelHandlerContext ctx) throws Exception {
+        ctx.flush();
+    }
+
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         ctx.fireExceptionCaught(cause);

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketServerProtocolHandler.java

@@ -110,7 +110,10 @@ public class WebSocketServerProtocolHandler extends WebSocketProtocolHandler {
      *            Server protocol configuration.
      */
     public WebSocketServerProtocolHandler(WebSocketServerProtocolConfig serverConfig) {
-        super(checkNotNull(serverConfig, "serverConfig").dropPongFrames());
+        super(checkNotNull(serverConfig, "serverConfig").dropPongFrames(),
+              serverConfig.sendCloseFrame(),
+              serverConfig.forceCloseTimeoutMillis()
+        );
         this.serverConfig = serverConfig;
     }
 
@@ -226,10 +229,6 @@ public class WebSocketServerProtocolHandler extends WebSocketProtocolHandler {
             cp.addBefore(ctx.name(), Utf8FrameValidator.class.getName(),
                     new Utf8FrameValidator());
         }
-        if (serverConfig.sendCloseFrame() != null) {
-            cp.addBefore(ctx.name(), WebSocketCloseFrameHandler.class.getName(),
-                new WebSocketCloseFrameHandler(serverConfig.sendCloseFrame(), serverConfig.forceCloseTimeoutMillis()));
-        }
     }
 
     @Override