import org.elasticsearch.transport.Transport; //导入依赖的package包/类
public static MockTransportService nettyFromThreadPool(Settings settings, ThreadPool threadPool, final Version version,
        ClusterSettings clusterSettings, boolean doHandshake) {
    NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(Collections.emptyList());
    Transport transport = new Netty4Transport(settings, threadPool, new NetworkService(settings, Collections.emptyList()),
        BigArrays.NON_RECYCLING_INSTANCE, namedWriteableRegistry, new NoneCircuitBreakerService()) {

        @Override
        protected Version executeHandshake(DiscoveryNode node, Channel channel, TimeValue timeout) throws IOException,
            InterruptedException {
            if (doHandshake) {
                return super.executeHandshake(node, channel, timeout);
            } else {
                return version.minimumCompatibilityVersion();
            }
        }

        @Override
        protected Version getCurrentVersion() {
            return version;
        }
    };
    MockTransportService mockTransportService =
        MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings);
    mockTransportService.start();
    return mockTransportService;
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
protected AbstractSearchAsyncAction(String name, Logger logger, SearchTransportService searchTransportService,
                                    Function<String, Transport.Connection> nodeIdToConnection,
                                    Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
                                    Executor executor, SearchRequest request,
                                    ActionListener<SearchResponse> listener, GroupShardsIterator shardsIts, long startTime,
                                    long clusterStateVersion, SearchTask task, SearchPhaseResults<Result> resultConsumer) {
    super(name, request, shardsIts, logger);
    this.startTime = startTime;
    this.logger = logger;
    this.searchTransportService = searchTransportService;
    this.executor = executor;
    this.request = request;
    this.task = task;
    this.listener = listener;
    this.nodeIdToConnection = nodeIdToConnection;
    this.clusterStateVersion = clusterStateVersion;
    this.concreteIndexBoosts = concreteIndexBoosts;
    this.aliasFilter = aliasFilter;
    this.results = resultConsumer;
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
/**
 * Returns a connection to the remote cluster. This connection might be a proxy connection that redirects internally to the
 * given node.
 */
Transport.Connection getConnection(DiscoveryNode remoteClusterNode) {
    DiscoveryNode discoveryNode = nodeSupplier.get();
    Transport.Connection connection = transportService.getConnection(discoveryNode);
    return new Transport.Connection() {
        @Override
        public DiscoveryNode getNode() {
            return remoteClusterNode;
        }

        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            connection.sendRequest(requestId, TransportActionProxy.getProxyAction(action),
                TransportActionProxy.wrapRequest(remoteClusterNode, request), options);
        }

        @Override
        public void close() throws IOException {
            assert false: "proxy connections must not be closed";
        }
    };
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
@Override
public void setUp() throws Exception {
    super.setUp();
    Settings settings = Settings.builder()
            .put(MapperService.INDEX_MAPPER_DYNAMIC_SETTING.getKey(), false)
            .build();
    clusterService = createClusterService(threadPool);
    Transport transport = new MockTcpTransport(settings, threadPool, BigArrays.NON_RECYCLING_INSTANCE,
            new NoneCircuitBreakerService(), new NamedWriteableRegistry(Collections.emptyList()),
            new NetworkService(settings, Collections.emptyList()));
    transportService = new TransportService(clusterService.getSettings(), transport, threadPool,
        TransportService.NOOP_TRANSPORT_INTERCEPTOR, x -> clusterService.localNode(), null);
    IndicesService indicesService = getInstanceFromNode(IndicesService.class);
    ShardStateAction shardStateAction = new ShardStateAction(settings, clusterService, transportService, null, null, threadPool);
    ActionFilters actionFilters = new ActionFilters(Collections.emptySet());
    IndexNameExpressionResolver indexNameExpressionResolver = new IndexNameExpressionResolver(settings);
    AutoCreateIndex autoCreateIndex = new AutoCreateIndex(settings, new ClusterSettings(settings,
            ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), indexNameExpressionResolver);
    UpdateHelper updateHelper = new UpdateHelper(settings, null);
    TransportShardBulkAction shardBulkAction = new TransportShardBulkAction(settings, transportService, clusterService,
            indicesService, threadPool, shardStateAction, null, updateHelper, actionFilters, indexNameExpressionResolver);
    transportBulkAction = new TransportBulkAction(settings, threadPool, transportService, clusterService,
            null, shardBulkAction, null, actionFilters, indexNameExpressionResolver, autoCreateIndex, System::currentTimeMillis);
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
@Override
public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
                                                      CircuitBreakerService circuitBreakerService,
                                                      NamedWriteableRegistry namedWriteableRegistry,
                                                      NetworkService networkService) {
    return Collections.singletonMap("exception-throwing",
        () -> new ExceptionThrowingNetty4Transport(settings, threadPool, networkService, bigArrays,
        namedWriteableRegistry, circuitBreakerService));
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
public static MockTransportService createNewService(Settings settings, Version version, ThreadPool threadPool,
        @Nullable ClusterSettings clusterSettings) {
    NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(ClusterModule.getNamedWriteables());
    final Transport transport = new MockTcpTransport(settings, threadPool, BigArrays.NON_RECYCLING_INSTANCE,
            new NoneCircuitBreakerService(), namedWriteableRegistry, new NetworkService(settings, Collections.emptyList()), version);
    return createNewService(settings, transport, version, threadPool, clusterSettings);
}
 

import org.elasticsearch.transport.Transport; //导入依赖的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,
        @Nullable ClusterSettings clusterSettings) {
    this(settings, transport, threadPool, interceptor, (boundAddress) ->
        DiscoveryNode.createLocal(settings, boundAddress.publishAddress(), settings.get(Node.NODE_NAME_SETTING.getKey(),
            UUIDs.randomBase64UUID())), clusterSettings);
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
/**
 * Clears the rule associated with the provided transport address.
 */
public void clearRule(TransportAddress transportAddress) {
    Transport transport = transport().transports.remove(transportAddress);
    if (transport instanceof ClearableTransport) {
        ((ClearableTransport) transport).clearRule();
    }
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
/**
 * simulates a disconnect by disconnecting from the underlying transport and throwing a
 * {@link ConnectTransportException}, due to a specific cause exception
 */
private void simulateDisconnect(DiscoveryNode node, Transport transport, String reason, @Nullable Throwable e) {
    if (transport.nodeConnected(node)) {
        // this a connected node, disconnecting from it will be up the exception
        transport.disconnectFromNode(node);
    } else {
        throw new ConnectTransportException(node, reason, e);
    }
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
private Transport getTransport(DiscoveryNode node) {
    Transport transport = transports.get(node.getAddress());
    if (transport != null) {
        return transport;
    }
    return this.transport;
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
public Transport getOriginalTransport() {
    Transport transport = transport();
    while (transport instanceof DelegateTransport) {
        transport = ((DelegateTransport) transport).transport;
    }
    return transport;
}
 

import org.elasticsearch.transport.Transport; //导入依赖的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.transport.Transport; //导入依赖的package包/类
public Supplier<Transport> getTransportSupplier() {
    final String name;
    if (TRANSPORT_TYPE_SETTING.exists(settings)) {
        name = TRANSPORT_TYPE_SETTING.get(settings);
    } else {
        name = TRANSPORT_DEFAULT_TYPE_SETTING.get(settings);
    }
    final Supplier<Transport> factory = transportFactories.get(name);
    if (factory == null) {
        throw new IllegalStateException("Unsupported transport.type [" + name + "]");
    }
    return factory;
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
/**
 * This method should be called if a search phase failed to ensure all relevant search contexts and resources are released.
 * this method will also notify the listener and sends back a failure to the user.
 *
 * @param exception the exception explaining or causing the phase failure
 */
private void raisePhaseFailure(SearchPhaseExecutionException exception) {
    results.getSuccessfulResults().forEach((entry) -> {
        try {
            Transport.Connection connection = nodeIdToConnection.apply(entry.shardTarget().getNodeId());
            sendReleaseSearchContext(entry.id(), connection);
        } catch (Exception inner) {
            inner.addSuppressed(exception);
            logger.trace("failed to release context", inner);
        }
    });
    listener.onFailure(exception);
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
private void executeFetch(final int shardIndex, final SearchShardTarget shardTarget,
                          final CountedCollector<FetchSearchResult> counter,
                          final ShardFetchSearchRequest fetchSearchRequest, final QuerySearchResult querySearchResult,
                          final Transport.Connection connection) {
    context.getSearchTransport().sendExecuteFetch(connection, fetchSearchRequest, context.getTask(),
        new ActionListener<FetchSearchResult>() {
            @Override
            public void onResponse(FetchSearchResult result) {
                counter.onResult(shardIndex, result, shardTarget);
            }

            @Override
            public void onFailure(Exception e) {
                try {
                    if (logger.isDebugEnabled()) {
                        logger.debug((Supplier<?>) () -> new ParameterizedMessage("[{}] Failed to execute fetch phase",
                            fetchSearchRequest.id()), e);
                    }
                    counter.onFailure(shardIndex, shardTarget, e);
                } finally {
                    // the search context might not be cleared on the node where the fetch was executed for example
                    // because the action was rejected by the thread pool. in this case we need to send a dedicated
                    // request to clear the search context.
                    releaseIrrelevantSearchContext(querySearchResult);
                }
            }
        });
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
/**
 * Releases shard targets that are not used in the docsIdsToLoad.
 */
private void releaseIrrelevantSearchContext(QuerySearchResult queryResult) {
    // we only release search context that we did not fetch from if we are not scrolling
    // and if it has at lease one hit that didn't make it to the global topDocs
    if (context.getRequest().scroll() == null && queryResult.hasHits()) {
        try {
            Transport.Connection connection = context.getConnection(queryResult.shardTarget().getNodeId());
            context.sendReleaseSearchContext(queryResult.id(), connection);
        } catch (Exception e) {
            context.getLogger().trace("failed to release context", e);
        }
    }
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
SniffClusterStateResponseHandler(TransportService transportService, Transport.Connection connection,
                                 ActionListener<Void> listener, Iterator<DiscoveryNode> seedNodes,
                                 CancellableThreads cancellableThreads) {
    this.transportService = transportService;
    this.connection = connection;
    this.listener = listener;
    this.seedNodes = seedNodes;
    this.cancellableThreads = cancellableThreads;
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
public void sendFreeContext(Transport.Connection connection, final long contextId, SearchRequest request) {
    transportService.sendRequest(connection, FREE_CONTEXT_ACTION_NAME, new SearchFreeContextRequest(request, contextId),
        TransportRequestOptions.EMPTY, new ActionListenerResponseHandler<>(new ActionListener<SearchFreeContextResponse>() {
            @Override
            public void onResponse(SearchFreeContextResponse response) {
                // no need to respond if it was freed or not
            }

            @Override
            public void onFailure(Exception e) {

            }
        }, SearchFreeContextResponse::new));
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
public void sendExecuteQuery(Transport.Connection connection, final ShardSearchTransportRequest request, SearchTask task,
                             final ActionListener<QuerySearchResultProvider> listener) {
    // we optimize this and expect a QueryFetchSearchResult if we only have a single shard in the search request
    // this used to be the QUERY_AND_FETCH which doesn't exists anymore.
    final boolean fetchDocuments = request.numberOfShards() == 1;
    Supplier<QuerySearchResultProvider> supplier = fetchDocuments ? QueryFetchSearchResult::new : QuerySearchResult::new;
    if (connection.getVersion().onOrBefore(Version.V_5_3_0_UNRELEASED) && fetchDocuments) {
        // TODO this BWC layer can be removed once this is back-ported to 5.3
        transportService.sendChildRequest(connection, QUERY_FETCH_ACTION_NAME, request, task,
            new ActionListenerResponseHandler<>(listener, supplier));
    } else {
        transportService.sendChildRequest(connection, QUERY_ACTION_NAME, request, task,
            new ActionListenerResponseHandler<>(listener, supplier));
    }
}
 

import org.elasticsearch.transport.Transport; //导入依赖的package包/类
Function<String, Transport.Connection> processRemoteShards(Map<String, ClusterSearchShardsResponse> searchShardsResponses,
                                                                   List<ShardIterator> remoteShardIterators,
                                                                   Map<String, AliasFilter> aliasFilterMap) {
    Map<String, Supplier<Transport.Connection>> nodeToCluster = new HashMap<>();
    for (Map.Entry<String, ClusterSearchShardsResponse> entry : searchShardsResponses.entrySet()) {
        String clusterName = entry.getKey();
        ClusterSearchShardsResponse searchShardsResponse = entry.getValue();
        for (DiscoveryNode remoteNode : searchShardsResponse.getNodes()) {
            nodeToCluster.put(remoteNode.getId(), () -> getConnection(remoteNode, clusterName));
        }
        Map<String, AliasFilter> indicesAndFilters = searchShardsResponse.getIndicesAndFilters();
        for (ClusterSearchShardsGroup clusterSearchShardsGroup : searchShardsResponse.getGroups()) {
            //add the cluster name to the remote index names for indices disambiguation
            //this ends up in the hits returned with the search response
            ShardId shardId = clusterSearchShardsGroup.getShardId();
            Index remoteIndex = shardId.getIndex();
            Index index = new Index(clusterName + REMOTE_CLUSTER_INDEX_SEPARATOR + remoteIndex.getName(), remoteIndex.getUUID());
            ShardIterator shardIterator = new PlainShardIterator(new ShardId(index, shardId.getId()),
                Arrays.asList(clusterSearchShardsGroup.getShards()));
            remoteShardIterators.add(shardIterator);
            AliasFilter aliasFilter;
            if (indicesAndFilters == null) {
                aliasFilter = new AliasFilter(null, Strings.EMPTY_ARRAY);
            } else {
                aliasFilter = indicesAndFilters.get(shardId.getIndexName());
                assert aliasFilter != null;
            }
            // here we have to map the filters to the UUID since from now on we use the uuid for the lookup
            aliasFilterMap.put(remoteIndex.getUUID(), aliasFilter);
        }
    }
    return (nodeId) -> {
        Supplier<Transport.Connection> supplier = nodeToCluster.get(nodeId);
        if (supplier == null) {
            throw new IllegalArgumentException("unknown remote node: " + nodeId);
        }
        return supplier.get();
    };
}