Introduce http and tcp server channels

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java

@@ -29,8 +29,6 @@
 import org.elasticsearch.http.HttpPipelinedRequest;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
-import static org.elasticsearch.http.netty4.Netty4HttpServerTransport.HTTP_CHANNEL_KEY;
-
 @ChannelHandler.Sharable
 class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelinedRequest<FullHttpRequest>> {
 
@@ -42,7 +40,7 @@ class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelined
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest<FullHttpRequest> msg) throws Exception {
-        Netty4HttpChannel channel = ctx.channel().attr(HTTP_CHANNEL_KEY).get();
+        Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         FullHttpRequest request = msg.getRequest();
 
         try {
@@ -77,12 +75,11 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest<Full
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         Netty4Utils.maybeDie(cause);
-        Netty4HttpChannel httpChannel = ctx.channel().attr(HTTP_CHANNEL_KEY).get();
+        Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         if (cause instanceof Error) {
-            serverTransport.onException(httpChannel, new Exception(cause));
+            serverTransport.onException(channel, new Exception(cause));
         } else {
-            serverTransport.onException(httpChannel, (Exception) cause);
+            serverTransport.onException(channel, (Exception) cause);
         }
     }
-
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerChannel.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.http.netty4;
+
+import io.netty.channel.Channel;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.concurrent.CompletableContext;
+import org.elasticsearch.http.HttpServerChannel;
+import org.elasticsearch.transport.netty4.Netty4Utils;
+
+import java.net.InetSocketAddress;
+
+public class Netty4HttpServerChannel implements HttpServerChannel {
+
+    private final Channel channel;
+    private final CompletableContext<Void> closeContext = new CompletableContext<>();
+
+    Netty4HttpServerChannel(Channel channel) {
+        this.channel = channel;
+        this.channel.closeFuture().addListener(f -> {
+            if (f.isSuccess()) {
+                closeContext.complete(null);
+            } else {
+                Throwable cause = f.cause();
+                if (cause instanceof Error) {
+                    Netty4Utils.maybeDie(cause);
+                    closeContext.completeExceptionally(new Exception(cause));
+                } else {
+                    closeContext.completeExceptionally((Exception) cause);
+                }
+            }
+        });
+    }
+
+    @Override
+    public InetSocketAddress getLocalAddress() {
+        return (InetSocketAddress) channel.localAddress();
+    }
+
+    @Override
+    public void addCloseListener(ActionListener<Void> listener) {
+        closeContext.addListener(ActionListener.toBiConsumer(listener));
+    }
+
+    @Override
+    public boolean isOpen() {
+        return channel.isOpen();
+    }
+
+    @Override
+    public void close() {
+        channel.close();
+    }
+
+    @Override
+    public String toString() {
+        return "Netty4HttpChannel{localAddress=" + getLocalAddress() + "}";
+    }
+}

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -23,6 +23,7 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
@@ -42,37 +43,29 @@
 import io.netty.util.AttributeKey;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.network.CloseableChannel;
-import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.AbstractHttpServerTransport;
-import org.elasticsearch.http.BindHttpException;
 import org.elasticsearch.http.HttpChannel;
 import org.elasticsearch.http.HttpHandlingSettings;
-import org.elasticsearch.http.HttpStats;
+import org.elasticsearch.http.HttpServerChannel;
 import org.elasticsearch.http.netty4.cors.Netty4CorsConfig;
 import org.elasticsearch.http.netty4.cors.Netty4CorsConfigBuilder;
 import org.elasticsearch.http.netty4.cors.Netty4CorsHandler;
 import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
-import java.io.IOException;
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
 
 import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory;
@@ -154,21 +147,13 @@ public class Netty4HttpServerTransport extends AbstractHttpServerTransport {
 
     private final int pipeliningMaxEvents;
 
-    private final boolean tcpNoDelay;
-    private final boolean tcpKeepAlive;
-    private final boolean reuseAddress;
-
-    private final ByteSizeValue tcpSendBufferSize;
-    private final ByteSizeValue tcpReceiveBufferSize;
     private final RecvByteBufAllocator recvByteBufAllocator;
     private final int readTimeoutMillis;
 
     private final int maxCompositeBufferComponents;
 
     protected volatile ServerBootstrap serverBootstrap;
 
-    protected final List<Channel> serverChannels = new ArrayList<>();
-
     private final Netty4CorsConfig corsConfig;
 
     public Netty4HttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays, ThreadPool threadPool,
@@ -184,11 +169,6 @@ public Netty4HttpServerTransport(Settings settings, NetworkService networkServic
         this.maxCompositeBufferComponents = SETTING_HTTP_NETTY_MAX_COMPOSITE_BUFFER_COMPONENTS.get(settings);
         this.workerCount = SETTING_HTTP_WORKER_COUNT.get(settings);
 
-        this.tcpNoDelay = SETTING_HTTP_TCP_NO_DELAY.get(settings);
-        this.tcpKeepAlive = SETTING_HTTP_TCP_KEEP_ALIVE.get(settings);
-        this.reuseAddress = SETTING_HTTP_TCP_REUSE_ADDRESS.get(settings);
-        this.tcpSendBufferSize = SETTING_HTTP_TCP_SEND_BUFFER_SIZE.get(settings);
-        this.tcpReceiveBufferSize = SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE.get(settings);
         this.readTimeoutMillis = Math.toIntExact(SETTING_HTTP_READ_TIMEOUT.get(settings).getMillis());
 
         ByteSizeValue receivePredictor = SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_SIZE.get(settings);
@@ -217,6 +197,7 @@ protected void doStart() {
             serverBootstrap.channel(NioServerSocketChannel.class);
 
             serverBootstrap.childHandler(configureServerChannelHandler());
+            serverBootstrap.handler(new ServerChannelExceptionHandler(this));
 
             serverBootstrap.childOption(ChannelOption.TCP_NODELAY, SETTING_HTTP_TCP_NO_DELAY.get(settings));
             serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, SETTING_HTTP_TCP_KEEP_ALIVE.get(settings));
@@ -238,10 +219,7 @@ protected void doStart() {
             serverBootstrap.option(ChannelOption.SO_REUSEADDR, reuseAddress);
             serverBootstrap.childOption(ChannelOption.SO_REUSEADDR, reuseAddress);
 
-            this.boundAddress = createBoundHttpAddress();
-            if (logger.isInfoEnabled()) {
-                logger.info("{}", boundAddress);
-            }
+            bindServer();
             success = true;
         } finally {
             if (success == false) {
@@ -284,78 +262,29 @@ static Netty4CorsConfig buildCorsConfig(Settings settings) {
     }
 
     @Override
-    protected TransportAddress bindAddress(final InetAddress hostAddress) {
-        final AtomicReference<Exception> lastException = new AtomicReference<>();
-        final AtomicReference<InetSocketAddress> boundSocket = new AtomicReference<>();
-        boolean success = port.iterate(portNumber -> {
-            try {
-                synchronized (serverChannels) {
-                    ChannelFuture future = serverBootstrap.bind(new InetSocketAddress(hostAddress, portNumber)).sync();
-                    serverChannels.add(future.channel());
-                    boundSocket.set((InetSocketAddress) future.channel().localAddress());
-                }
-            } catch (Exception e) {
-                lastException.set(e);
-                return false;
-            }
-            return true;
-        });
-        if (!success) {
-            throw new BindHttpException("Failed to bind to [" + port.getPortRangeString() + "]", lastException.get());
-        }
-
-        if (logger.isDebugEnabled()) {
-            logger.debug("Bound http to address {{}}", NetworkAddress.format(boundSocket.get()));
-        }
-        return new TransportAddress(boundSocket.get());
+    protected HttpServerChannel bind(InetSocketAddress socketAddress) throws Exception {
+        ChannelFuture future = serverBootstrap.bind(socketAddress).sync();
+        Channel channel = future.channel();
+        Netty4HttpServerChannel httpServerChannel = new Netty4HttpServerChannel(channel);
+        channel.attr(HTTP_SERVER_CHANNEL_KEY).set(httpServerChannel);
+        return httpServerChannel;
     }
 
     @Override
-    protected void doStop() {
-        synchronized (serverChannels) {
-            if (!serverChannels.isEmpty()) {
-                try {
-                    Netty4Utils.closeChannels(serverChannels);
-                } catch (IOException e) {
-                    logger.trace("exception while closing channels", e);
-                } finally {
-                    serverChannels.clear();
-                }
-            }
-        }
-
-        // TODO: Move all of channel closing to abstract class once server channels are handled
-        try {
-            CloseableChannel.closeChannels(new ArrayList<>(httpChannels), true);
-        } catch (Exception e) {
-            logger.warn("unexpected exception while closing http channels", e);
-        }
-        httpChannels.clear();
-
-
-
+    protected void stopInternal() {
         if (serverBootstrap != null) {
             serverBootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
             serverBootstrap = null;
         }
     }
 
-    @Override
-    protected void doClose() {
-    }
-
-    @Override
-    public HttpStats stats() {
-        return new HttpStats(httpChannels.size(), totalChannelsAccepted.get());
-    }
-
     @Override
     protected void onException(HttpChannel channel, Exception cause) {
         if (cause instanceof ReadTimeoutException) {
             if (logger.isTraceEnabled()) {
                 logger.trace("Http read timeout {}", channel);
             }
-            CloseableChannel.closeChannel(channel);;
+            CloseableChannel.closeChannel(channel);
         } else {
             super.onException(channel, cause);
         }
@@ -366,6 +295,7 @@ public ChannelHandler configureServerChannelHandler() {
     }
 
     static final AttributeKey<Netty4HttpChannel> HTTP_CHANNEL_KEY = AttributeKey.newInstance("es-http-channel");
+    static final AttributeKey<Netty4HttpServerChannel> HTTP_SERVER_CHANNEL_KEY = AttributeKey.newInstance("es-http-server-channel");
 
     protected static class HttpChannelHandler extends ChannelInitializer<Channel> {
 
@@ -413,4 +343,24 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
         }
     }
 
+    @ChannelHandler.Sharable
+    private static class ServerChannelExceptionHandler extends ChannelHandlerAdapter {
+
+        private final Netty4HttpServerTransport transport;
+
+        private ServerChannelExceptionHandler(Netty4HttpServerTransport transport) {
+            this.transport = transport;
+        }
+
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+            Netty4Utils.maybeDie(cause);
+            Netty4HttpServerChannel httpServerChannel = ctx.channel().attr(HTTP_SERVER_CHANNEL_KEY).get();
+            if (cause instanceof Error) {
+                transport.onServerException(httpServerChannel, new Exception(cause));
+            } else {
+                transport.onServerException(httpServerChannel, (Exception) cause);
+            }
+        }
+    }
 }

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageChannelHandler.java

@@ -24,6 +24,8 @@
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.Attribute;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.transport.TcpHeader;
 import org.elasticsearch.transport.Transports;
@@ -36,11 +38,9 @@
 final class Netty4MessageChannelHandler extends ChannelDuplexHandler {
 
     private final Netty4Transport transport;
-    private final String profileName;
 
-    Netty4MessageChannelHandler(Netty4Transport transport, String profileName) {
+    Netty4MessageChannelHandler(Netty4Transport transport) {
         this.transport = transport;
-        this.profileName = profileName;
     }
 
     @Override
@@ -58,7 +58,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             // netty always copies a buffer, either in NioWorker in its read handler, where it copies to a fresh
             // buffer, or in the cumulative buffer, which is cleaned each time so it could be bigger than the actual size
             BytesReference reference = Netty4Utils.toBytesReference(buffer, remainingMessageSize);
-            Attribute<NettyTcpChannel> channelAttribute = channel.attr(Netty4Transport.CHANNEL_KEY);
+            Attribute<Netty4TcpChannel> channelAttribute = channel.attr(Netty4Transport.CHANNEL_KEY);
             transport.messageReceived(reference, channelAttribute.get());
         } finally {
             // Set the expected position of the buffer, no matter what happened
@@ -69,7 +69,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         Netty4Utils.maybeDie(cause);
-        transport.exceptionCaught(ctx, cause);
+        final Throwable unwrapped = ExceptionsHelper.unwrap(cause, ElasticsearchException.class);
+        final Throwable newCause = unwrapped != null ? unwrapped : cause;
+        Netty4TcpChannel tcpChannel = ctx.channel().attr(Netty4Transport.CHANNEL_KEY).get();
+        if (newCause instanceof Error) {
+            transport.onException(tcpChannel, new Exception(newCause));
+        } else {
+            transport.onException(tcpChannel, (Exception) newCause);
+        }
     }
-
 }