import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public void testThatNettyHttpServerDoesNotSupportPipelining() throws Exception {
    ensureGreen();
    String[] requests = new String[] {"/", "/_nodes/stats", "/", "/_cluster/state", "/", "/_nodes", "/"};

    HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
    TransportAddress[] boundAddresses = httpServerTransport.boundAddress().boundAddresses();
    TransportAddress transportAddress = (TransportAddress) randomFrom(boundAddresses);

    try (Netty4HttpClient nettyHttpClient = new Netty4HttpClient()) {
        Collection<FullHttpResponse> responses = nettyHttpClient.get(transportAddress.address(), requests);
        assertThat(responses, hasSize(requests.length));

        List<String> opaqueIds = new ArrayList<>(Netty4HttpClient.returnOpaqueIds(responses));

        assertResponsesOutOfOrder(opaqueIds);
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public void testThatHttpPipeliningWorksWhenEnabled() throws Exception {
    final Settings settings = Settings.builder()
        .put("http.pipelining", true)
        .put("http.port", "0")
        .build();
    try (HttpServerTransport httpServerTransport = new CustomNettyHttpServerTransport(settings)) {
        httpServerTransport.start();
        final TransportAddress transportAddress = randomFrom(httpServerTransport.boundAddress().boundAddresses());

        final int numberOfRequests = randomIntBetween(4, 16);
        final List<String> requests = new ArrayList<>(numberOfRequests);
        for (int i = 0; i < numberOfRequests; i++) {
            if (rarely()) {
                requests.add("/slow/" + i);
            } else {
                requests.add("/" + i);
            }
        }

        try (Netty4HttpClient nettyHttpClient = new Netty4HttpClient()) {
            Collection<FullHttpResponse> responses = nettyHttpClient.get(transportAddress.address(), requests.toArray(new String[]{}));
            Collection<String> responseBodies = Netty4HttpClient.returnHttpResponseBodies(responses);
            assertThat(responseBodies, contains(requests.toArray()));
        }
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public void testDoesNotLimitExcludedRequests() throws Exception {
    ensureGreen();

    @SuppressWarnings("unchecked")
    Tuple<String, CharSequence>[] requestUris = new Tuple[1500];
    for (int i = 0; i < requestUris.length; i++) {
        requestUris[i] = Tuple.tuple("/_cluster/settings",
            "{ \"transient\": {\"search.default_search_timeout\": \"40s\" } }");
    }

    HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
    TransportAddress transportAddress = (TransportAddress) randomFrom(httpServerTransport.boundAddress
        ().boundAddresses());

    try (Netty4HttpClient nettyHttpClient = new Netty4HttpClient()) {
        Collection<FullHttpResponse> responses = nettyHttpClient.put(transportAddress.address(), requestUris);
        assertThat(responses, hasSize(requestUris.length));
        assertAllInExpectedStatus(responses, HttpResponseStatus.OK);
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
NodeService(Settings settings, ThreadPool threadPool, MonitorService monitorService, Discovery discovery,
                   TransportService transportService, IndicesService indicesService, PluginsService pluginService,
                   CircuitBreakerService circuitBreakerService, ScriptService scriptService,
                   @Nullable HttpServerTransport httpServerTransport, IngestService ingestService, ClusterService clusterService,
                   SettingsFilter settingsFilter) {
    super(settings);
    this.threadPool = threadPool;
    this.monitorService = monitorService;
    this.transportService = transportService;
    this.indicesService = indicesService;
    this.discovery = discovery;
    this.pluginService = pluginService;
    this.circuitBreakerService = circuitBreakerService;
    this.httpServerTransport = httpServerTransport;
    this.ingestService = ingestService;
    this.settingsFilter = settingsFilter;
    this.scriptService = scriptService;
    clusterService.addStateApplier(ingestService.getPipelineStore());
    clusterService.addStateApplier(ingestService.getPipelineExecutionService());
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
@Before
public void setup() {
    Settings settings = Settings.EMPTY;
    circuitBreakerService = new HierarchyCircuitBreakerService(
        Settings.builder()
            .put(HierarchyCircuitBreakerService.IN_FLIGHT_REQUESTS_CIRCUIT_BREAKER_LIMIT_SETTING.getKey(), BREAKER_LIMIT)
            .build(),
        new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS));
    // we can do this here only because we know that we don't adjust breaker settings dynamically in the test
    inFlightRequestsBreaker = circuitBreakerService.getBreaker(CircuitBreaker.IN_FLIGHT_REQUESTS);

    HttpServerTransport httpServerTransport = new TestHttpServerTransport();
    restController = new RestController(settings, Collections.emptySet(), null, null, circuitBreakerService);
    restController.registerHandler(RestRequest.Method.GET, "/",
        (request, channel, client) -> channel.sendResponse(
            new BytesRestResponse(RestStatus.OK, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY)));
    restController.registerHandler(RestRequest.Method.GET, "/error", (request, channel, client) -> {
        throw new IllegalArgumentException("test error");
    });

    httpServerTransport.start();
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
@Override
public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
        CircuitBreakerService circuitBreakerService, NamedWriteableRegistry namedWriteableRegistry,
        NamedXContentRegistry xContentRegistry, NetworkService networkService, Dispatcher dispatcher) {
    
    Map<String, Supplier<HttpServerTransport>> httpTransports = new HashMap<String, Supplier<HttpServerTransport>>(1);

    if(!disabled) {
        if (!client && httpSSLEnabled && !tribeNodeClient) {
            
            final ValidatingDispatcher validatingDispatcher = new ValidatingDispatcher(threadPool.getThreadContext(), dispatcher, 
                    settings, configPath, evaluateSslExceptionHandler());
            final SearchGuardHttpServerTransport sghst = new SearchGuardHttpServerTransport(settings, networkService, bigArrays, 
                    threadPool, sgks, evaluateSslExceptionHandler(), xContentRegistry, validatingDispatcher);
            
            httpTransports.put("com.floragunn.searchguard.http.SearchGuardHttpServerTransport", 
                    () -> sghst);
        } else if (!client && !tribeNodeClient) {
            httpTransports.put("com.floragunn.searchguard.http.SearchGuardHttpServerTransport", 
                    () -> new SearchGuardNonSslHttpServerTransport(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher));
        }
    }
    return httpTransports;
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
@Override
public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
        CircuitBreakerService circuitBreakerService, NamedWriteableRegistry namedWriteableRegistry,
        NamedXContentRegistry xContentRegistry, NetworkService networkService, Dispatcher dispatcher) {
    
    final Map<String, Supplier<HttpServerTransport>> httpTransports = new HashMap<String, Supplier<HttpServerTransport>>(1);
    if (!client && httpSSLEnabled) {
        
        final ValidatingDispatcher validatingDispatcher = new ValidatingDispatcher(threadPool.getThreadContext(), dispatcher, settings, configPath, NOOP_SSL_EXCEPTION_HANDLER);
        final SearchGuardSSLNettyHttpServerTransport sgsnht = new SearchGuardSSLNettyHttpServerTransport(settings, networkService, bigArrays, threadPool, sgks, xContentRegistry, validatingDispatcher, NOOP_SSL_EXCEPTION_HANDLER);
        
        httpTransports.put("com.floragunn.searchguard.ssl.http.netty.SearchGuardSSLNettyHttpServerTransport", () -> sgsnht);
        
    }
    return httpTransports;
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
@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(NETTY_HTTP_TRANSPORT_NAME,
        () -> new Netty4HttpServerTransport(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher));
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public void testThatNettyHttpServerSupportsPipelining() throws Exception {
    String[] requests = new String[]{"/", "/_nodes/stats", "/", "/_cluster/state", "/"};

    HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
    TransportAddress[] boundAddresses = httpServerTransport.boundAddress().boundAddresses();
    TransportAddress transportAddress = (TransportAddress) randomFrom(boundAddresses);

    try (Netty4HttpClient nettyHttpClient = new Netty4HttpClient()) {
        Collection<FullHttpResponse> responses = nettyHttpClient.get(transportAddress.address(), requests);
        assertThat(responses, hasSize(5));

        Collection<String> opaqueIds = Netty4HttpClient.returnOpaqueIds(responses);
        assertOpaqueIdsInOrder(opaqueIds);
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public void testLimitsInFlightRequests() throws Exception {
    ensureGreen();

    // we use the limit size as a (very) rough indication on how many requests we should sent to hit the limit
    int numRequests = LIMIT.bytesAsInt() / 100;

    StringBuilder bulkRequest = new StringBuilder();
    for (int i = 0; i < numRequests; i++) {
        bulkRequest.append("{\"index\": {}}");
        bulkRequest.append(System.lineSeparator());
        bulkRequest.append("{ \"field\" : \"value\" }");
        bulkRequest.append(System.lineSeparator());
    }

    @SuppressWarnings("unchecked")
    Tuple<String, CharSequence>[] requests = new Tuple[150];
    for (int i = 0; i < requests.length; i++) {
        requests[i] = Tuple.tuple("/index/type/_bulk", bulkRequest);
    }

    HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
    TransportAddress transportAddress = (TransportAddress) randomFrom(httpServerTransport.boundAddress
        ().boundAddresses());

    try (Netty4HttpClient nettyHttpClient = new Netty4HttpClient()) {
        Collection<FullHttpResponse> singleResponse = nettyHttpClient.post(transportAddress.address(), requests[0]);
        assertThat(singleResponse, hasSize(1));
        assertAtLeastOnceExpectedStatus(singleResponse, HttpResponseStatus.OK);

        Collection<FullHttpResponse> multipleResponses = nettyHttpClient.post(transportAddress.address(), requests);
        assertThat(multipleResponses, hasSize(requests.length));
        assertAtLeastOnceExpectedStatus(multipleResponses, HttpResponseStatus.SERVICE_UNAVAILABLE);
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
@Override
public InetSocketAddress[] httpAddresses() {
    List<InetSocketAddress> addresses = new ArrayList<>();
    for (HttpServerTransport httpServerTransport : getInstances(HttpServerTransport.class)) {
        addresses.add(httpServerTransport.boundAddress().publishAddress().address());
    }
    return addresses.toArray(new InetSocketAddress[addresses.size()]);
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
/** Adds an http transport implementation that can be selected by setting {@link #HTTP_TYPE_KEY}. */
// TODO: we need another name than "http transport"....so confusing with transportClient...
private void registerHttpTransport(String key, Supplier<HttpServerTransport> factory) {
    if (transportClient) {
        throw new IllegalArgumentException("Cannot register http transport " + key + " for transport client");
    }
    if (transportHttpFactories.putIfAbsent(key, factory) != null) {
        throw new IllegalArgumentException("transport for name: " + key + " is already registered");
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public Supplier<HttpServerTransport> getHttpServerTransportSupplier() {
    final String name;
    if (HTTP_TYPE_SETTING.exists(settings)) {
        name = HTTP_TYPE_SETTING.get(settings);
    } else {
        name = HTTP_DEFAULT_TYPE_SETTING.get(settings);
    }
    final Supplier<HttpServerTransport> factory = transportHttpFactories.get(name);
    if (factory == null) {
        throw new IllegalStateException("Unsupported http.type [" + name + "]");
    }
    return factory;
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
private Node stop() {
    if (!lifecycle.moveToStopped()) {
        return this;
    }
    Logger logger = Loggers.getLogger(Node.class, NODE_NAME_SETTING.get(settings));
    logger.info("stopping ...");

    injector.getInstance(TribeService.class).stop();
    injector.getInstance(ResourceWatcherService.class).stop();
    if (NetworkModule.HTTP_ENABLED.get(settings)) {
        injector.getInstance(HttpServerTransport.class).stop();
    }

    injector.getInstance(SnapshotsService.class).stop();
    injector.getInstance(SnapshotShardsService.class).stop();
    // stop any changes happening as a result of cluster state changes
    injector.getInstance(IndicesClusterStateService.class).stop();
    // close discovery early to not react to pings anymore.
    // This can confuse other nodes and delay things - mostly if we're the master and we're running tests.
    injector.getInstance(Discovery.class).stop();
    // we close indices first, so operations won't be allowed on it
    injector.getInstance(RoutingService.class).stop();
    injector.getInstance(ClusterService.class).stop();
    injector.getInstance(NodeConnectionsService.class).stop();
    injector.getInstance(MonitorService.class).stop();
    injector.getInstance(GatewayService.class).stop();
    injector.getInstance(SearchService.class).stop();
    injector.getInstance(TransportService.class).stop();
    injector.getInstance(SearchTransportService.class).stop();

    pluginLifecycleComponents.forEach(LifecycleComponent::stop);
    // we should stop this last since it waits for resources to get released
    // if we had scroll searchers etc or recovery going on we wait for to finish.
    injector.getInstance(IndicesService.class).stop();
    logger.info("stopped");

    return this;
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
/**
 * Returns a map of {@link HttpServerTransport} suppliers.
 * See {@link org.elasticsearch.common.network.NetworkModule#HTTP_TYPE_SETTING} to configure a specific implementation.
 */
default Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
                                                                     CircuitBreakerService circuitBreakerService,
                                                                     NamedWriteableRegistry namedWriteableRegistry,
                                                                     NamedXContentRegistry xContentRegistry,
                                                                     NetworkService networkService,
                                                                     HttpServerTransport.Dispatcher dispatcher) {
    return Collections.emptyMap();
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
private int getPortOfRunningNode() {
    TransportAddress transportAddress = internalCluster().getInstance(HttpServerTransport.class).boundAddress().publishAddress();
    if (transportAddress instanceof InetSocketTransportAddress) {
        return ((InetSocketTransportAddress) transportAddress).address().getPort();
    }
    throw new ElasticsearchException("Could not find running tcp port");
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
public void testThatHttpPipeliningCanBeDisabled() throws Exception {
    final Settings settings = Settings.builder()
        .put("http.pipelining", false)
        .put("http.port", "0")
        .build();
    try (HttpServerTransport httpServerTransport = new CustomNettyHttpServerTransport(settings)) {
        httpServerTransport.start();
        final TransportAddress transportAddress = randomFrom(httpServerTransport.boundAddress().boundAddresses());

        final int numberOfRequests = randomIntBetween(4, 16);
        final Set<Integer> slowIds = new HashSet<>();
        final List<String> requests = new ArrayList<>(numberOfRequests);
        for (int i = 0; i < numberOfRequests; i++) {
            if (rarely()) {
                requests.add("/slow/" + i);
                slowIds.add(i);
            } else {
                requests.add("/" + i);
            }
        }

        try (Netty4HttpClient nettyHttpClient = new Netty4HttpClient()) {
            Collection<FullHttpResponse> responses = nettyHttpClient.get(transportAddress.address(), requests.toArray(new String[]{}));
            List<String> responseBodies = new ArrayList<>(Netty4HttpClient.returnHttpResponseBodies(responses));
            // we can not be sure about the order of the responses, but the slow ones should come last
            assertThat(responseBodies, hasSize(numberOfRequests));
            for (int i = 0; i < numberOfRequests - slowIds.size(); i++) {
                assertThat(responseBodies.get(i), matches("/\\d+"));
            }

            final Set<Integer> ids = new HashSet<>();
            for (int i = 0; i < slowIds.size(); i++) {
                final String response = responseBodies.get(numberOfRequests - slowIds.size() + i);
                assertThat(response, matches("/slow/\\d+" ));
                assertTrue(ids.add(Integer.parseInt(response.split("/")[2])));
            }

            assertThat(slowIds, equalTo(ids));
        }
    }
}
 

import org.elasticsearch.http.HttpServerTransport; //导入依赖的package包/类
/**
 * Start the node. If the node is already started, this method is no-op.
 */
public Node start() {
    if (!lifecycle.moveToStarted()) {
        return this;
    }

    ESLogger logger = Loggers.getLogger(Node.class, settings.get("name"));
    logger.info("starting ...");
    // hack around dependency injection problem (for now...)
    injector.getInstance(Discovery.class).setRoutingService(injector.getInstance(RoutingService.class));
    for (Class<? extends LifecycleComponent> plugin : pluginsService.nodeServices()) {
        injector.getInstance(plugin).start();
    }

    injector.getInstance(MappingUpdatedAction.class).setClient(client);
    injector.getInstance(IndicesService.class).start();
    injector.getInstance(IndexingMemoryController.class).start();
    injector.getInstance(IndicesClusterStateService.class).start();
    injector.getInstance(IndicesTTLService.class).start();
    injector.getInstance(SnapshotsService.class).start();
    injector.getInstance(SnapshotShardsService.class).start();
    injector.getInstance(RoutingService.class).start();
    injector.getInstance(SearchService.class).start();
    injector.getInstance(MonitorService.class).start();
    injector.getInstance(RestController.class).start();

    // TODO hack around circular dependencies problems
    injector.getInstance(GatewayAllocator.class).setReallocation(injector.getInstance(ClusterService.class), injector.getInstance(RoutingService.class));

    injector.getInstance(ResourceWatcherService.class).start();
    injector.getInstance(GatewayService.class).start();
    injector.getInstance(TenantManagementService.class).start();

    // Start the transport service now so the publish address will be added to the local disco node in ClusterService
    TransportService transportService = injector.getInstance(TransportService.class);
    transportService.start();
    injector.getInstance(ClusterService.class).start();

    // start after cluster service so the local disco is known
    DiscoveryService discoService = injector.getInstance(DiscoveryService.class).start();


    transportService.acceptIncomingRequests();
    discoService.joinClusterAndWaitForInitialState();

    if (settings.getAsBoolean("http.enabled", true)) {
        injector.getInstance(HttpServer.class).start();
    }
    injector.getInstance(TribeService.class).start();
    if (settings.getAsBoolean("node.portsfile", false)) {
        if (settings.getAsBoolean("http.enabled", true)) {
            HttpServerTransport http = injector.getInstance(HttpServerTransport.class);
            writePortsFile("http", http.boundAddress());
        }
        TransportService transport = injector.getInstance(TransportService.class);
        writePortsFile("transport", transport.boundAddress());
    }
    logger.info("started");

    return this;
}