import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
/**
 * Adds a rule that will cause every send request to fail, and each new connect since the rule
 * is added to fail as well.
 */
public void addFailToSendNoConnectRule(TransportAddress transportAddress) {
    addDelegate(transportAddress, new DelegateTransport(original) {

        @Override
        public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfile,
                                  CheckedBiConsumer<Connection, ConnectionProfile, IOException> connectionValidator)
            throws ConnectTransportException {
            if (original.nodeConnected(node) == false) {
                // connecting to an already connected node is a no-op
                throw new ConnectTransportException(node, "DISCONNECT: simulated");
            }
        }

        @Override
        protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                                   TransportRequestOptions options) throws IOException {
            simulateDisconnect(connection, original, "DISCONNECT: simulated");
        }
    });
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
/**
 * Adds a rule that will cause ignores each send request, simulating an unresponsive node
 * and failing to connect once the rule was added.
 */
public void addUnresponsiveRule(TransportAddress transportAddress) {
    addDelegate(transportAddress, new DelegateTransport(original) {

        @Override
        public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfile,
                                  CheckedBiConsumer<Connection, ConnectionProfile, IOException> connectionValidator)
            throws ConnectTransportException {
            if (original.nodeConnected(node) == false) {
                // connecting to an already connected node is a no-op
                throw new ConnectTransportException(node, "UNRESPONSIVE: simulated");
            }
        }

        @Override
        protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                                   TransportRequestOptions options) throws IOException {
            // don't send anything, the receiving node is unresponsive
        }
    });
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
    return new Connection() {
        @Override
        public DiscoveryNode getNode() {
            return node;
        }

        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            requests.put(requestId, Tuple.tuple(node, action));
            capturedRequests.add(new CapturedRequest(node, requestId, action, request));
        }

        @Override
        public void close() throws IOException {

        }
    };
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
public RemoteRecoveryTargetHandler(long recoveryId, ShardId shardId, String targetAllocationId, TransportService transportService,
                                   DiscoveryNode targetNode, RecoverySettings recoverySettings, Consumer<Long> onSourceThrottle) {
    this.targetAllocationId = targetAllocationId;
    this.transportService = transportService;


    this.recoveryId = recoveryId;
    this.shardId = shardId;
    this.targetNode = targetNode;
    this.recoverySettings = recoverySettings;
    this.onSourceThrottle = onSourceThrottle;
    this.translogOpsRequestOptions = TransportRequestOptions.builder()
            .withCompress(true)
            .withType(TransportRequestOptions.Type.RECOVERY)
            .withTimeout(recoverySettings.internalActionLongTimeout())
            .build();
    this.fileChunkRequestOptions = TransportRequestOptions.builder()
            .withCompress(false)  // lucene files are already compressed and therefore compressing this won't really help much so
            // we are saving the cpu for other things
            .withType(TransportRequestOptions.Type.RECOVERY)
            .withTimeout(recoverySettings.internalActionTimeout())
            .build();

}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public Connection getConnection(DiscoveryNode node) {
    return new Connection() {
        @Override
        public DiscoveryNode getNode() {
            return node;
        }

        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {

        }

        @Override
        public void close() throws IOException {

        }
    };
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request, TransportRequestOptions options) throws IOException {
    if (action.equals(PeerRecoveryTargetService.Actions.FILE_CHUNK)) {
        RecoveryFileChunkRequest chunkRequest = (RecoveryFileChunkRequest) request;
        if (chunkRequest.name().startsWith(IndexFileNames.SEGMENTS)) {
            // corrupting the segments_N files in order to make sure future recovery re-send files
            logger.debug("corrupting [{}] to {}. file name: [{}]", action, connection.getNode(), chunkRequest.name());
            assert chunkRequest.content().toBytesRef().bytes == chunkRequest.content().toBytesRef().bytes : "no internal reference!!";
            byte[] array = chunkRequest.content().toBytesRef().bytes;
            array[0] = (byte) ~array[0]; // flip one byte in the content
            corruptionCount.countDown();
        }
        super.sendRequest(connection, requestId, action, request, options);
    } else {
        super.sendRequest(connection, requestId, action, request, options);
    }
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
protected void prepareTargetForTranslog() {
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("{} recovery [phase1] to {}: prepare remote engine for translog", request.shardId(), request.targetNode());
    final long startEngineStart = stopWatch.totalTime().millis();
    cancellableThreads.execute(new Interruptable() {
        @Override
        public void run() throws InterruptedException {
            // Send a request preparing the new shard's translog to receive
            // operations. This ensures the shard engine is started and disables
            // garbage collection (not the JVM's GC!) of tombstone deletes
            transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.PREPARE_TRANSLOG,
                    new RecoveryPrepareForTranslogOperationsRequest(request.recoveryId(), request.shardId(), 0),
                    TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(), EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
        }
    });

    stopWatch.stop();

    response.startTime = stopWatch.totalTime().millis() - startEngineStart;
    logger.trace("{} recovery [phase1] to {}: remote engine start took [{}]",
            request.shardId(), request.targetNode(), stopWatch.totalTime());
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
protected void prepareTargetForTranslog(final Translog.View translogView) {
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("{} recovery [phase1] to {}: prepare remote engine for translog", request.shardId(), request.targetNode());
    final long startEngineStart = stopWatch.totalTime().millis();
    cancellableThreads.execute(new Interruptable() {
        @Override
        public void run() throws InterruptedException {
            // Send a request preparing the new shard's translog to receive
            // operations. This ensures the shard engine is started and disables
            // garbage collection (not the JVM's GC!) of tombstone deletes
            transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.PREPARE_TRANSLOG,
                    new RecoveryPrepareForTranslogOperationsRequest(request.recoveryId(), request.shardId(), translogView.totalOperations()),
                    TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(), EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
        }
    });

    stopWatch.stop();

    response.startTime = stopWatch.totalTime().millis() - startEngineStart;
    logger.trace("{} recovery [phase1] to {}: remote engine start took [{}]",
            request.shardId(), request.targetNode(), stopWatch.totalTime());
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public Connection getConnection(DiscoveryNode node) {
    return new FilteredConnection(transport.getConnection(node)) {
        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            DelegateTransport.this.sendRequest(connection, requestId, action, request, options);
        }
    };
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
    return new FilteredConnection(transport.openConnection(node, profile)) {
        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            DelegateTransport.this.sendRequest(connection, requestId, action, request, options);
        }
    };
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
protected void traceRequestSent(DiscoveryNode node, long requestId, String action, TransportRequestOptions options) {
    super.traceRequestSent(node, requestId, action, options);
    for (Tracer tracer : activeTracers) {
        tracer.requestSent(node, requestId, action, options);
    }
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public void prepareForTranslogOperations(int totalTranslogOps, long maxUnsafeAutoIdTimestamp) throws IOException {
    transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.PREPARE_TRANSLOG,
            new RecoveryPrepareForTranslogOperationsRequest(recoveryId, shardId, totalTranslogOps, maxUnsafeAutoIdTimestamp),
            TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
            EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public void finalizeRecovery(final long globalCheckpoint) {
    transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.FINALIZE,
        new RecoveryFinalizeRecoveryRequest(recoveryId, shardId, globalCheckpoint),
        TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionLongTimeout()).build(),
        EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public void ensureClusterStateVersion(long clusterStateVersion) {
    transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.WAIT_CLUSTERSTATE,
        new RecoveryWaitForClusterStateRequest(recoveryId, shardId, clusterStateVersion),
        TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionLongTimeout()).build(),
        EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public void receiveFileInfo(List<String> phase1FileNames, List<Long> phase1FileSizes, List<String> phase1ExistingFileNames,
                            List<Long> phase1ExistingFileSizes, int totalTranslogOps) {

    RecoveryFilesInfoRequest recoveryInfoFilesRequest = new RecoveryFilesInfoRequest(recoveryId, shardId,
            phase1FileNames, phase1FileSizes, phase1ExistingFileNames, phase1ExistingFileSizes, totalTranslogOps);
    transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.FILES_INFO, recoveryInfoFilesRequest,
            TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
            EmptyTransportResponseHandler.INSTANCE_SAME).txGet();

}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
@Override
public void cleanFiles(int totalTranslogOps, Store.MetadataSnapshot sourceMetaData) throws IOException {
    transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.CLEAN_FILES,
            new RecoveryCleanFilesRequest(recoveryId, shardId, sourceMetaData, totalTranslogOps),
            TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
            EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
private void sendCommitToNode(final DiscoveryNode node, final ClusterState clusterState, final SendingController sendingController) {
    try {
        logger.trace("sending commit for cluster state (uuid: [{}], version [{}]) to [{}]",
            clusterState.stateUUID(), clusterState.version(), node);
        TransportRequestOptions options = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).build();
        // no need to put a timeout on the options here, because we want the response to eventually be received
        // and not log an error if it arrives after the timeout
        transportService.sendRequest(node, COMMIT_ACTION_NAME,
                new CommitClusterStateRequest(clusterState.stateUUID()),
                options,
                new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {

                    @Override
                    public void handleResponse(TransportResponse.Empty response) {
                        if (sendingController.getPublishingTimedOut()) {
                            logger.debug("node {} responded to cluster state commit [{}]", node, clusterState.version());
                        }
                        sendingController.getPublishResponseHandler().onResponse(node);
                    }

                    @Override
                    public void handleException(TransportException exp) {
                        logger.debug((org.apache.logging.log4j.util.Supplier<?>) () ->
                            new ParameterizedMessage("failed to commit cluster state (uuid [{}], version [{}]) to {}",
                                clusterState.stateUUID(), clusterState.version(), node), exp);
                        sendingController.getPublishResponseHandler().onFailure(node, exp);
                    }
                });
    } catch (Exception t) {
        logger.warn((org.apache.logging.log4j.util.Supplier<?>) () ->
            new ParameterizedMessage("error sending cluster state commit (uuid [{}], version [{}]) to {}",
                clusterState.stateUUID(), clusterState.version(), node), t);
        sendingController.getPublishResponseHandler().onFailure(node, t);
    }
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的package包/类
/**
 * Creates a new {@link RemoteClusterConnection}
 * @param settings the nodes settings object
 * @param clusterAlias the configured alias of the cluster to connect to
 * @param seedNodes a list of seed nodes to discover eligible nodes from
 * @param transportService the local nodes transport service
 * @param maxNumRemoteConnections the maximum number of connections to the remote cluster
 * @param nodePredicate a predicate to filter eligible remote nodes to connect to
 */
RemoteClusterConnection(Settings settings, String clusterAlias, List<DiscoveryNode> seedNodes,
                        TransportService transportService, int maxNumRemoteConnections, Predicate<DiscoveryNode> nodePredicate) {
    super(settings);
    this.transportService = transportService;
    this.maxNumRemoteConnections = maxNumRemoteConnections;
    this.nodePredicate = nodePredicate;
    this.clusterAlias = clusterAlias;
    ConnectionProfile.Builder builder = new ConnectionProfile.Builder();
    builder.setConnectTimeout(TcpTransport.TCP_CONNECT_TIMEOUT.get(settings));
    builder.setHandshakeTimeout(TcpTransport.TCP_CONNECT_TIMEOUT.get(settings));
    builder.addConnections(6, TransportRequestOptions.Type.REG, TransportRequestOptions.Type.PING); // TODO make this configurable?
    builder.addConnections(0, // we don't want this to be used for anything else but search
        TransportRequestOptions.Type.BULK,
        TransportRequestOptions.Type.STATE,
        TransportRequestOptions.Type.RECOVERY);
    remoteProfile = builder.build();
    nodeSupplier = new Supplier<DiscoveryNode>() {
        private volatile Iterator<DiscoveryNode> current;
        @Override
        public DiscoveryNode get() {
            if (current == null || current.hasNext() == false) {
                current = connectedNodes.iterator();
                if (current.hasNext() == false) {
                    throw new IllegalStateException("No node available for cluster: " + clusterAlias + " nodes: " + connectedNodes);
                }
            }
            return current.next();
        }
    };
    this.seedNodes = Collections.unmodifiableList(seedNodes);
    this.connectHandler = new ConnectHandler();
    transportService.addConnectionListener(this);
}
 

import org.elasticsearch.transport.TransportRequestOptions; //导入依赖的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.TransportRequestOptions; //导入依赖的package包/类
@Override
public TransportRequestOptions transportOptions(Settings settings) {
    return TransportRequestOptions.builder()
            .withType(TransportRequestOptions.Type.BULK)
            .withCompress(settings.getAsBoolean("action.bulk.compress", true)
            ).build();
}