import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
/**
 * Build the service.
 *
 * @param clusterSettings if non null the the {@linkplain TransportService} will register with the {@link ClusterSettings} for settings
 *        updates for {@link #TRACE_LOG_EXCLUDE_SETTING} and {@link #TRACE_LOG_INCLUDE_SETTING}.
 */
public TransportService(Settings settings, Transport transport, ThreadPool threadPool, TransportInterceptor transportInterceptor,
                        Function<BoundTransportAddress, DiscoveryNode> localNodeFactory, @Nullable ClusterSettings clusterSettings) {
    super(settings);
    this.transport = transport;
    this.threadPool = threadPool;
    this.localNodeFactory = localNodeFactory;
    this.clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings);
    setTracerLogInclude(TRACE_LOG_INCLUDE_SETTING.get(settings));
    setTracerLogExclude(TRACE_LOG_EXCLUDE_SETTING.get(settings));
    tracerLog = Loggers.getLogger(logger, ".tracer");
    adapter = createAdapter();
    taskManager = createTaskManager();
    this.interceptor = transportInterceptor;
    this.asyncSender = interceptor.interceptSender(this::sendRequestInternal);
    if (clusterSettings != null) {
        clusterSettings.addSettingsUpdateConsumer(TRACE_LOG_INCLUDE_SETTING, this::setTracerLogInclude);
        clusterSettings.addSettingsUpdateConsumer(TRACE_LOG_EXCLUDE_SETTING, this::setTracerLogExclude);
    }
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
@Override
protected void doStart() {
    adapter.rxMetric.clear();
    adapter.txMetric.clear();
    transport.transportServiceAdapter(adapter);
    transport.start();

    if (transport.boundAddress() != null && logger.isInfoEnabled()) {
        logger.info("{}", transport.boundAddress());
        for (Map.Entry<String, BoundTransportAddress> entry : transport.profileBoundAddresses().entrySet()) {
            logger.info("profile [{}]: {}", entry.getKey(), entry.getValue());
        }
    }
    localNode = localNodeFactory.apply(transport.boundAddress());
    registerRequestHandler(
        HANDSHAKE_ACTION_NAME,
        () -> HandshakeRequest.INSTANCE,
        ThreadPool.Names.SAME,
        false, false,
        (request, channel) -> channel.sendResponse(
                new HandshakeResponse(localNode, clusterName, localNode.getVersion())));
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
    builder.startObject(Fields.TRANSPORT);
    builder.array(Fields.BOUND_ADDRESS, (Object[]) address.boundAddresses());
    builder.field(Fields.PUBLISH_ADDRESS, address.publishAddress().toString());
    builder.startObject(Fields.PROFILES);
    if (profileAddresses != null && profileAddresses.size() > 0) {
        for (Map.Entry<String, BoundTransportAddress> entry : profileAddresses.entrySet()) {
            builder.startObject(entry.getKey());
            builder.array(Fields.BOUND_ADDRESS, (Object[]) entry.getValue().boundAddresses());
            builder.field(Fields.PUBLISH_ADDRESS, entry.getValue().publishAddress().toString());
            builder.endObject();
        }
    }
    builder.endObject();
    builder.endObject();
    return builder;
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public void testEnforceLimitsWhenBoundToNonLocalAddress() {
    final List<TransportAddress> transportAddresses = new ArrayList<>();
    final TransportAddress nonLocalTransportAddress = buildNewFakeTransportAddress();
    transportAddresses.add(nonLocalTransportAddress);

    for (int i = 0; i < randomIntBetween(0, 7); i++) {
        final TransportAddress randomTransportAddress = randomBoolean() ? buildNewFakeTransportAddress() :
            new TransportAddress(InetAddress.getLoopbackAddress(), i);
        transportAddresses.add(randomTransportAddress);
    }

    final TransportAddress publishAddress = randomBoolean() ? buildNewFakeTransportAddress() :
        new TransportAddress(InetAddress.getLoopbackAddress(), 0);

    final BoundTransportAddress boundTransportAddress = mock(BoundTransportAddress.class);
    Collections.shuffle(transportAddresses, random());
    when(boundTransportAddress.boundAddresses()).thenReturn(transportAddresses.toArray(new TransportAddress[0]));
    when(boundTransportAddress.publishAddress()).thenReturn(publishAddress);

    assertTrue(BootstrapChecks.enforceLimits(boundTransportAddress));
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
@Before
public void createTransportSvc() {
    MockTcpTransport transport =
        new MockTcpTransport(Settings.EMPTY,
                                threadPool,
                                BigArrays.NON_RECYCLING_INSTANCE,
                                new NoneCircuitBreakerService(),
                                new NamedWriteableRegistry(Collections.emptyList()),
                                new NetworkService(Settings.EMPTY, Collections.emptyList())) {
            @Override
            public BoundTransportAddress boundAddress() {
                return new BoundTransportAddress(
                    new TransportAddress[]{new TransportAddress(InetAddress.getLoopbackAddress(), 9300)},
                    new TransportAddress(InetAddress.getLoopbackAddress(), 9300)
                );
            }
        };
    transportService = new MockTransportService(Settings.EMPTY, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR,
            null);
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
@Override
public HttpInfo info() {
    BoundTransportAddress boundTransportAddress = boundAddress();
    if (boundTransportAddress == null) {
        return null;
    }
    return new HttpInfo(boundTransportAddress, maxContentLength.getBytes());
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public void testDifferentPorts() throws Exception {
    if (!NetworkUtils.SUPPORTS_V6) {
        return;
    }
    logger.info("--> starting a node on ipv4 only");
    Settings ipv4Settings = Settings.builder().put("network.host", "127.0.0.1").build();
    String ipv4OnlyNode = internalCluster().startNode(ipv4Settings); // should bind 127.0.0.1:XYZ

    logger.info("--> starting a node on ipv4 and ipv6");
    Settings bothSettings = Settings.builder().put("network.host", "_local_").build();
    internalCluster().startNode(bothSettings); // should bind [::1]:XYZ and 127.0.0.1:XYZ+1

    logger.info("--> waiting for the cluster to declare itself stable");
    ensureStableCluster(2); // fails if port of publish address does not match corresponding bound address

    logger.info("--> checking if boundAddress matching publishAddress has same port");
    NodesInfoResponse nodesInfoResponse = client().admin().cluster().prepareNodesInfo().get();
    for (NodeInfo nodeInfo : nodesInfoResponse.getNodes()) {
        BoundTransportAddress boundTransportAddress = nodeInfo.getTransport().getAddress();
        if (nodeInfo.getNode().getName().equals(ipv4OnlyNode)) {
            assertThat(boundTransportAddress.boundAddresses().length, equalTo(1));
            assertThat(boundTransportAddress.boundAddresses()[0].getPort(), equalTo(boundTransportAddress.publishAddress().getPort()));
        } else {
            assertThat(boundTransportAddress.boundAddresses().length, greaterThan(1));
            for (TransportAddress boundAddress : boundTransportAddress.boundAddresses()) {
                assertThat(boundAddress, instanceOf(TransportAddress.class));
                TransportAddress inetBoundAddress = (TransportAddress) boundAddress;
                if (inetBoundAddress.address().getAddress() instanceof Inet4Address) {
                    // IPv4 address is preferred publish address for _local_
                    assertThat(inetBoundAddress.getPort(), equalTo(boundTransportAddress.publishAddress().getPort()));
                }
            }
        }
    }
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
/**
 * Build the service.
 *
 * @param clusterSettings if non null the the {@linkplain TransportService} will register with the {@link ClusterSettings} for settings
 *        updates for {@link #TRACE_LOG_EXCLUDE_SETTING} and {@link #TRACE_LOG_INCLUDE_SETTING}.
 */
public MockTransportService(Settings settings, Transport transport, ThreadPool threadPool, TransportInterceptor interceptor,
                            Function<BoundTransportAddress, DiscoveryNode> localNodeFactory,
                            @Nullable ClusterSettings clusterSettings) {
    super(settings, new LookupTestTransport(transport), threadPool, interceptor, localNodeFactory, clusterSettings);
    this.original = transport;
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public static TransportAddress[] extractTransportAddresses(TransportService transportService) {
    HashSet<TransportAddress> transportAddresses = new HashSet<>();
    BoundTransportAddress boundTransportAddress = transportService.boundAddress();
    transportAddresses.addAll(Arrays.asList(boundTransportAddress.boundAddresses()));
    transportAddresses.add(boundTransportAddress.publishAddress());
    return transportAddresses.toArray(new TransportAddress[transportAddresses.size()]);
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
@Override
protected TransportService newTransportService(Settings settings, Transport transport, ThreadPool threadPool,
                                               TransportInterceptor interceptor,
                                               Function<BoundTransportAddress, DiscoveryNode> localNodeFactory,
                                               ClusterSettings clusterSettings) {
    // we use the MockTransportService.TestPlugin class as a marker to create a network
    // module with this MockNetworkService. NetworkService is such an integral part of the systme
    // we don't allow to plug it in from plugins or anything. this is a test-only override and
    // can't be done in a production env.
    if (getPluginsService().filterPlugins(MockTransportService.TestPlugin.class).isEmpty()) {
        return super.newTransportService(settings, transport, threadPool, interceptor, localNodeFactory, clusterSettings);
    } else {
        return new MockTransportService(settings, transport, threadPool, interceptor, localNodeFactory, clusterSettings);
    }
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public TransportInfo info() {
    BoundTransportAddress boundTransportAddress = boundAddress();
    if (boundTransportAddress == null) {
        return null;
    }
    return new TransportInfo(boundTransportAddress, transport.profileBoundAddresses());
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
private BoundTransportAddress createBoundTransportAddress(String name, Settings profileSettings,
                                                          List<InetSocketAddress> boundAddresses) {
    String[] boundAddressesHostStrings = new String[boundAddresses.size()];
    TransportAddress[] transportBoundAddresses = new TransportAddress[boundAddresses.size()];
    for (int i = 0; i < boundAddresses.size(); i++) {
        InetSocketAddress boundAddress = boundAddresses.get(i);
        boundAddressesHostStrings[i] = boundAddress.getHostString();
        transportBoundAddresses[i] = new TransportAddress(boundAddress);
    }

    final String[] publishHosts;
    if (TransportSettings.DEFAULT_PROFILE.equals(name)) {
        publishHosts = TransportSettings.PUBLISH_HOST.get(settings).toArray(Strings.EMPTY_ARRAY);
    } else {
        publishHosts = profileSettings.getAsArray("publish_host", boundAddressesHostStrings);
    }

    final InetAddress publishInetAddress;
    try {
        publishInetAddress = networkService.resolvePublishHostAddresses(publishHosts);
    } catch (Exception e) {
        throw new BindTransportException("Failed to resolve publish address", e);
    }

    final int publishPort = resolvePublishPort(name, settings, profileSettings, boundAddresses, publishInetAddress);
    final TransportAddress publishAddress = new TransportAddress(new InetSocketAddress(publishInetAddress, publishPort));
    return new BoundTransportAddress(transportBoundAddresses, publishAddress);
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public TransportInfo(StreamInput in) throws IOException {
    address = BoundTransportAddress.readBoundTransportAddress(in);
    int size = in.readVInt();
    if (size > 0) {
        profileAddresses = new HashMap<>(size);
        for (int i = 0; i < size; i++) {
            String key = in.readString();
            BoundTransportAddress value = BoundTransportAddress.readBoundTransportAddress(in);
            profileAddresses.put(key, value);
        }
    }
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
@Override
public void writeTo(StreamOutput out) throws IOException {
    address.writeTo(out);
    if (profileAddresses != null) {
        out.writeVInt(profileAddresses.size());
    } else {
        out.writeVInt(0);
    }
    if (profileAddresses != null && profileAddresses.size() > 0) {
        for (Map.Entry<String, BoundTransportAddress> entry : profileAddresses.entrySet()) {
            out.writeString(entry.getKey());
            entry.getValue().writeTo(out);
        }
    }
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
/**
 * Executes the bootstrap checks if the node has the transport protocol bound to a non-loopback interface.
 *
 * @param settings              the current node settings
 * @param boundTransportAddress the node network bindings
 */
static void check(final Settings settings, final BoundTransportAddress boundTransportAddress, List<BootstrapCheck> additionalChecks)
    throws NodeValidationException {
    final List<BootstrapCheck> builtInChecks = checks(settings);
    final List<BootstrapCheck> combinedChecks = new ArrayList<>(builtInChecks);
    combinedChecks.addAll(additionalChecks);
    check(
            enforceLimits(boundTransportAddress),
            Collections.unmodifiableList(combinedChecks),
            Node.NODE_NAME_SETTING.get(settings));
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
/**
 * Tests if the checks should be enforced.
 *
 * @param boundTransportAddress the node network bindings
 * @return {@code true} if the checks should be enforced
 */
static boolean enforceLimits(BoundTransportAddress boundTransportAddress) {
    Predicate<TransportAddress> isLoopbackOrLinkLocalAddress = t -> t.address().getAddress().isLinkLocalAddress()
        || t.address().getAddress().isLoopbackAddress();
    return !(Arrays.stream(boundTransportAddress.boundAddresses()).allMatch(isLoopbackOrLinkLocalAddress) &&
            isLoopbackOrLinkLocalAddress.test(boundTransportAddress.publishAddress()));
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public void testPortLimit() throws InterruptedException {
    final NetworkService networkService = new NetworkService(Settings.EMPTY, Collections.emptyList());
    final Transport transport = new MockTcpTransport(
        Settings.EMPTY,
        threadPool,
        BigArrays.NON_RECYCLING_INSTANCE,
        new NoneCircuitBreakerService(),
        new NamedWriteableRegistry(Collections.emptyList()),
        networkService,
        Version.CURRENT) {

        @Override
        public BoundTransportAddress boundAddress() {
            return new BoundTransportAddress(
                new TransportAddress[]{new TransportAddress(InetAddress.getLoopbackAddress(), 9500)},
                new TransportAddress(InetAddress.getLoopbackAddress(), 9500)
            );
        }
    };
    closeables.push(transport);
    final TransportService transportService =
        new TransportService(Settings.EMPTY, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, x -> null, null);
    closeables.push(transportService);
    final int limitPortCounts = randomIntBetween(1, 10);
    final List<DiscoveryNode> discoveryNodes = TestUnicastZenPing.resolveHostsLists(
        executorService,
        logger,
        Collections.singletonList("127.0.0.1"),
        limitPortCounts,
        transportService,
        "test_",
        TimeValue.timeValueSeconds(1));
    assertThat(discoveryNodes, hasSize(limitPortCounts));
    final Set<Integer> ports = new HashSet<>();
    for (final DiscoveryNode discoveryNode : discoveryNodes) {
        assertTrue(discoveryNode.getAddress().address().getAddress().isLoopbackAddress());
        ports.add(discoveryNode.getAddress().getPort());
    }
    assertThat(ports, equalTo(IntStream.range(9300, 9300 + limitPortCounts).mapToObj(m -> m).collect(Collectors.toSet())));
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public void testInvalidHosts() throws InterruptedException {
    final Logger logger = mock(Logger.class);
    final NetworkService networkService = new NetworkService(Settings.EMPTY, Collections.emptyList());
    final Transport transport = new MockTcpTransport(
        Settings.EMPTY,
        threadPool,
        BigArrays.NON_RECYCLING_INSTANCE,
        new NoneCircuitBreakerService(),
        new NamedWriteableRegistry(Collections.emptyList()),
        networkService,
        Version.CURRENT) {
        @Override
        public BoundTransportAddress boundAddress() {
            return new BoundTransportAddress(
                new TransportAddress[]{new TransportAddress(InetAddress.getLoopbackAddress(), 9300)},
                new TransportAddress(InetAddress.getLoopbackAddress(), 9300)
            );
        }
    };
    closeables.push(transport);

    final TransportService transportService =
        new TransportService(Settings.EMPTY, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, x -> null, null);
    closeables.push(transportService);
    final List<DiscoveryNode> discoveryNodes = TestUnicastZenPing.resolveHostsLists(
        executorService,
        logger,
        Arrays.asList("127.0.0.1:9300:9300", "127.0.0.1:9301"),
        1,
        transportService,
        "test_",
        TimeValue.timeValueSeconds(1));
    assertThat(discoveryNodes, hasSize(1)); // only one of the two is valid and will be used
    assertThat(discoveryNodes.get(0).getAddress().getAddress(), equalTo("127.0.0.1"));
    assertThat(discoveryNodes.get(0).getAddress().getPort(), equalTo(9301));
    verify(logger).warn(eq("failed to resolve host [127.0.0.1:9300:9300]"), Matchers.any(ExecutionException.class));
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public TestNode(String name, ThreadPool threadPool, Settings settings) {
    final Function<BoundTransportAddress, DiscoveryNode> boundTransportAddressDiscoveryNodeFunction =
        address -> {
         discoveryNode.set(new DiscoveryNode(name, address.publishAddress(), emptyMap(), emptySet(), Version.CURRENT));
         return discoveryNode.get();
        };
    transportService = new TransportService(settings,
            new MockTcpTransport(settings, threadPool, BigArrays.NON_RECYCLING_INSTANCE, new NoneCircuitBreakerService(),
                new NamedWriteableRegistry(ClusterModule.getNamedWriteables()),
                new NetworkService(settings, Collections.emptyList())),
            threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, boundTransportAddressDiscoveryNodeFunction, null) {
        @Override
        protected TaskManager createTaskManager() {
            if (MockTaskManager.USE_MOCK_TASK_MANAGER_SETTING.get(settings)) {
                return new MockTaskManager(settings);
            } else {
                return super.createTaskManager();
            }
        }
    };
    transportService.start();
    clusterService = createClusterService(threadPool, discoveryNode.get());
    clusterService.addStateApplier(transportService.getTaskManager());
    IndexNameExpressionResolver indexNameExpressionResolver = new IndexNameExpressionResolver(settings);
    ActionFilters actionFilters = new ActionFilters(emptySet());
    transportListTasksAction = new TransportListTasksAction(settings, threadPool, clusterService, transportService,
            actionFilters, indexNameExpressionResolver);
    transportCancelTasksAction = new TransportCancelTasksAction(settings, threadPool, clusterService,
            transportService, actionFilters, indexNameExpressionResolver);
    transportService.acceptIncomingRequests();
}
 

import org.elasticsearch.common.transport.BoundTransportAddress; //导入依赖的package包/类
public void testNonProductionMode() throws NodeValidationException {
    // nothing should happen since we are in non-production mode
    final List<TransportAddress> transportAddresses = new ArrayList<>();
    for (int i = 0; i < randomIntBetween(1, 8); i++) {
        TransportAddress localTransportAddress = new TransportAddress(InetAddress.getLoopbackAddress(), i);
        transportAddresses.add(localTransportAddress);
    }

    TransportAddress publishAddress = new TransportAddress(InetAddress.getLoopbackAddress(), 0);
    BoundTransportAddress boundTransportAddress = mock(BoundTransportAddress.class);
    when(boundTransportAddress.boundAddresses()).thenReturn(transportAddresses.toArray(new TransportAddress[0]));
    when(boundTransportAddress.publishAddress()).thenReturn(publishAddress);
    BootstrapChecks.check(Settings.EMPTY, boundTransportAddress, Collections.emptyList());
}