[grid] Closing input pipeline when client disconnects

This is needed when a client sends a request, and we
reply right away (e.g. session not found), but we do
not read the whole request content.

If we do not read the whole request content, Netty
will wait until it does, and then it blocks. Which
is not ideal since no more requests can be processed.

Now, we close the pipeline when the client disconnects.

Fixes #10485
Fixes SeleniumHQ/docker-selenium#1439

Author: diemol

java/src/org/openqa/selenium/events/zeromq/UnboundZmqEventBus.java

@@ -231,7 +231,6 @@ public void run() {
               EventName eventName = new EventName(new String(socket.recv(), UTF_8));
               // Processing only events we are listening to
               if (!listeners.containsKey(eventName)) {
-//                LOG.log(Level.SEVERE, "Ignoring {0}", eventName);
                 continue;
               }
 

java/src/org/openqa/selenium/netty/server/RequestConverter.java

@@ -61,6 +61,7 @@ class RequestConverter extends SimpleChannelInboundHandler<HttpObject> {
 
   private static final Logger LOG = Logger.getLogger(RequestConverter.class.getName());
   private static final ExecutorService EXECUTOR = Executors.newSingleThreadExecutor();
+  private static final ExecutorService SHUTDOWN_EXECUTOR = Executors.newSingleThreadExecutor();
   private static final List<io.netty.handler.codec.http.HttpMethod> supportedMethods =
     Arrays.asList(DELETE, GET, POST, OPTIONS);
   private volatile PipedOutputStream out;
@@ -132,6 +133,19 @@ protected void channelRead0(
     }
   }
 
+  @Override
+  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+    LOG.fine("Closing input pipe, channel became inactive.");
+    SHUTDOWN_EXECUTOR.submit(() -> {
+      try {
+        out.close();
+      } catch (IOException e) {
+        throw new UncheckedIOException(e);
+      }
+    });
+    super.channelInactive(ctx);
+  }
+
   private HttpRequest createRequest(
     ChannelHandlerContext ctx,
     io.netty.handler.codec.http.HttpRequest nettyRequest) {