Add experimental nio-transport plugin with http

docs/build.gradle

@@ -46,6 +46,9 @@ project.rootProject.subprojects.findAll { it.parent.path == ':plugins' }.each {
   if (subproj.path.startsWith(':plugins:repository-')) {
     return
   }
+  if (subproj.path.startsWith(':plugins:transport-nio')) {
+    return
+  }
   subproj.afterEvaluate { // need to wait until the project has been configured
     integTestCluster {
       plugin subproj.path

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

@@ -39,7 +39,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-public class Netty4HttpRequest extends RestRequest {
+public final class Netty4HttpRequest extends RestRequest {
 
     private final FullHttpRequest request;
     private final Channel channel;

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

@@ -228,7 +228,7 @@ public Netty4HttpServerTransport(Settings settings, NetworkService networkServic
         this.compressionLevel = SETTING_HTTP_COMPRESSION_LEVEL.get(settings);
         this.pipelining = SETTING_PIPELINING.get(settings);
         this.pipeliningMaxEvents = SETTING_PIPELINING_MAX_EVENTS.get(settings);
-        this.corsConfig = buildCorsConfig(settings);
+        this.corsConfig = Netty4CorsConfig.buildCorsConfig(settings);
 
         // validate max content length
         if (maxContentLength.getBytes() > Integer.MAX_VALUE) {
@@ -351,40 +351,6 @@ static int resolvePublishPort(Settings settings, List<TransportAddress> boundAdd
         return publishPort;
     }
 
-    // package private for testing
-    static Netty4CorsConfig buildCorsConfig(Settings settings) {
-        if (SETTING_CORS_ENABLED.get(settings) == false) {
-            return Netty4CorsConfigBuilder.forOrigins().disable().build();
-        }
-        String origin = SETTING_CORS_ALLOW_ORIGIN.get(settings);
-        final Netty4CorsConfigBuilder builder;
-        if (Strings.isNullOrEmpty(origin)) {
-            builder = Netty4CorsConfigBuilder.forOrigins();
-        } else if (origin.equals(ANY_ORIGIN)) {
-            builder = Netty4CorsConfigBuilder.forAnyOrigin();
-        } else {
-            Pattern p = RestUtils.checkCorsSettingForRegex(origin);
-            if (p == null) {
-                builder = Netty4CorsConfigBuilder.forOrigins(RestUtils.corsSettingAsArray(origin));
-            } else {
-                builder = Netty4CorsConfigBuilder.forPattern(p);
-            }
-        }
-        if (SETTING_CORS_ALLOW_CREDENTIALS.get(settings)) {
-            builder.allowCredentials();
-        }
-        String[] strMethods = Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_METHODS.get(settings), ",");
-        HttpMethod[] methods = Arrays.asList(strMethods)
-            .stream()
-            .map(HttpMethod::valueOf)
-            .toArray(size -> new HttpMethod[size]);
-        return builder.allowedRequestMethods(methods)
-            .maxAge(SETTING_CORS_MAX_AGE.get(settings))
-            .allowedRequestHeaders(Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_HEADERS.get(settings), ","))
-            .shortCircuit()
-            .build();
-    }
-
     private TransportAddress bindAddress(final InetAddress hostAddress) {
         final AtomicReference<Exception> lastException = new AtomicReference<>();
         final AtomicReference<InetSocketAddress> boundSocket = new AtomicReference<>();

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

@@ -23,7 +23,11 @@
 import io.netty.handler.codec.http.EmptyHttpHeaders;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.rest.RestUtils;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.Map;
@@ -32,6 +36,14 @@
 import java.util.concurrent.Callable;
 import java.util.regex.Pattern;
 
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_CREDENTIALS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_HEADERS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_METHODS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_ORIGIN;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ENABLED;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_MAX_AGE;
+import static org.elasticsearch.http.netty4.cors.Netty4CorsHandler.ANY_ORIGIN;
+
 /**
  * Configuration for Cross-Origin Resource Sharing (CORS).
  *
@@ -232,4 +244,37 @@ public String toString() {
             ", preflightHeaders=" + preflightHeaders + ']';
     }
 
+    public static Netty4CorsConfig buildCorsConfig(Settings settings) {
+        if (SETTING_CORS_ENABLED.get(settings) == false) {
+            return Netty4CorsConfigBuilder.forOrigins().disable().build();
+        }
+        String origin = SETTING_CORS_ALLOW_ORIGIN.get(settings);
+        final Netty4CorsConfigBuilder builder;
+        if (Strings.isNullOrEmpty(origin)) {
+            builder = Netty4CorsConfigBuilder.forOrigins();
+        } else if (origin.equals(ANY_ORIGIN)) {
+            builder = Netty4CorsConfigBuilder.forAnyOrigin();
+        } else {
+            Pattern p = RestUtils.checkCorsSettingForRegex(origin);
+            if (p == null) {
+                builder = Netty4CorsConfigBuilder.forOrigins(RestUtils.corsSettingAsArray(origin));
+            } else {
+                builder = Netty4CorsConfigBuilder.forPattern(p);
+            }
+        }
+        if (SETTING_CORS_ALLOW_CREDENTIALS.get(settings)) {
+            builder.allowCredentials();
+        }
+        String[] strMethods = Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_METHODS.get(settings), ",");
+        HttpMethod[] methods = Arrays.asList(strMethods)
+            .stream()
+            .map(HttpMethod::valueOf)
+            .toArray(size -> new HttpMethod[size]);
+        return builder.allowedRequestMethods(methods)
+            .maxAge(SETTING_CORS_MAX_AGE.get(settings))
+            .allowedRequestHeaders(Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_HEADERS.get(settings), ","))
+            .shortCircuit()
+            .build();
+    }
+
 }

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

@@ -23,7 +23,7 @@
 import io.netty.handler.codec.http.FullHttpResponse;
 import io.netty.util.ReferenceCounted;
 
-class HttpPipelinedResponse implements Comparable<HttpPipelinedResponse>, ReferenceCounted {
+public class HttpPipelinedResponse implements Comparable<HttpPipelinedResponse>, ReferenceCounted {
 
     private final FullHttpResponse response;
     private final ChannelPromise promise;

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

@@ -27,7 +27,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 
-final class ByteBufBytesReference extends BytesReference {
+public final class ByteBufBytesReference extends BytesReference {
 
     private final ByteBuf buffer;
     private final int length;
@@ -65,7 +65,7 @@ public void writeTo(OutputStream os) throws IOException {
         buffer.getBytes(offset, os, length);
     }
 
-    ByteBuf toByteBuf() {
+    public ByteBuf toByteBuf() {
         return buffer.duplicate();
     }
 

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpServerTransportTests.java

@@ -117,7 +117,7 @@ public void testCorsConfig() {
                                       .put(SETTING_CORS_ALLOW_HEADERS.getKey(), collectionToDelimitedString(headers, ",", prefix, ""))
                                       .put(SETTING_CORS_ALLOW_CREDENTIALS.getKey(), true)
                                       .build();
-        final Netty4CorsConfig corsConfig = Netty4HttpServerTransport.buildCorsConfig(settings);
+        final Netty4CorsConfig corsConfig = Netty4CorsConfig.buildCorsConfig(settings);
         assertTrue(corsConfig.isAnyOriginSupported());
         assertEquals(headers, corsConfig.allowedRequestHeaders());
         assertEquals(methods, corsConfig.allowedRequestMethods().stream().map(HttpMethod::name).collect(Collectors.toSet()));
@@ -128,7 +128,7 @@ public void testCorsConfigWithDefaults() {
         final Set<String> headers = Strings.commaDelimitedListToSet(SETTING_CORS_ALLOW_HEADERS.getDefault(Settings.EMPTY));
         final long maxAge = SETTING_CORS_MAX_AGE.getDefault(Settings.EMPTY);
         final Settings settings = Settings.builder().put(SETTING_CORS_ENABLED.getKey(), true).build();
-        final Netty4CorsConfig corsConfig = Netty4HttpServerTransport.buildCorsConfig(settings);
+        final Netty4CorsConfig corsConfig = Netty4CorsConfig.buildCorsConfig(settings);
         assertFalse(corsConfig.isAnyOriginSupported());
         assertEquals(Collections.emptySet(), corsConfig.origins().get());
         assertEquals(headers, corsConfig.allowedRequestHeaders());

plugins/transport-nio/build.gradle

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+esplugin {
+  description 'The nio transport.'
+  classname 'org.elasticsearch.transport.nio.NioPlugin'
+}
+
+dependencyLicenses.enabled = false
+
+compileJava.options.compilerArgs << "-Xlint:-try"
+compileTestJava.options.compilerArgs << "-Xlint:-rawtypes,-unchecked"
+
+dependencies {
+  compile project(path: ':modules:transport-netty4', configuration: 'runtime')
+  compile project(path: ':test:framework', configuration: 'runtime')
+}

plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/NioNotEnabledBootstrapCheck.java

@@ -0,0 +1,32 @@
+/*
+ * 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.transport.nio;
+
+import org.elasticsearch.bootstrap.BootstrapCheck;
+import org.elasticsearch.bootstrap.BootstrapContext;
+
+public class NioNotEnabledBootstrapCheck implements BootstrapCheck {
+
+    @Override
+    public BootstrapCheckResult check(BootstrapContext context) {
+        return BootstrapCheckResult.failure("The transport-nio plugin is experimental and not ready for production usage. It should " +
+            "not be enabled in production.");
+    }
+}

plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/NioPlugin.java

@@ -0,0 +1,91 @@
+/*
+ * 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.transport.nio;
+
+import org.elasticsearch.bootstrap.BootstrapCheck;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.network.NetworkModule;
+import org.elasticsearch.common.network.NetworkService;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.http.HttpServerTransport;
+import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.plugins.NetworkPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.nio.http.NioHttpTransport;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+public class NioPlugin extends Plugin implements NetworkPlugin {
+
+    public static final String NIO_TRANSPORT_NAME = "nio-transport";
+    public static final String NIO_HTTP_TRANSPORT_NAME = "nio-http-transport";
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return Arrays.asList(
+            NioHttpTransport.NIO_HTTP_WORKER_COUNT,
+            NioHttpTransport.NIO_HTTP_ACCEPTOR_COUNT,
+            NioTransport.NIO_WORKER_COUNT,
+            NioTransport.NIO_ACCEPTOR_COUNT
+        );
+    }
+
+    @Override
+    public Settings additionalSettings() {
+        final Settings.Builder settingsBuilder = Settings.builder();
+        settingsBuilder.put(NetworkModule.TRANSPORT_TYPE_KEY, NIO_TRANSPORT_NAME);
+        settingsBuilder.put(NetworkModule.HTTP_TYPE_KEY, NIO_HTTP_TRANSPORT_NAME);
+        return settingsBuilder.build();
+    }
+
+    @Override
+    public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                          CircuitBreakerService circuitBreakerService,
+                                                          NamedWriteableRegistry namedWriteableRegistry,
+                                                          NetworkService networkService) {
+        return Collections.singletonMap(NIO_TRANSPORT_NAME, () -> new NioTransport(settings, threadPool, networkService, bigArrays,
+            namedWriteableRegistry, circuitBreakerService));
+    }
+
+    @Override
+    public List<BootstrapCheck> getBootstrapChecks() {
+        return Collections.singletonList(new NioNotEnabledBootstrapCheck());
+    }
+
+    @Override
+    public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                                        CircuitBreakerService circuitBreakerService,
+                                                                        NamedWriteableRegistry namedWriteableRegistry,
+                                                                        NamedXContentRegistry xContentRegistry,
+                                                                        NetworkService networkService,
+                                                                        HttpServerTransport.Dispatcher dispatcher) {
+        return Collections.singletonMap(NIO_HTTP_TRANSPORT_NAME,
+            () -> new NioHttpTransport(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher));
+    }
+}