import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
@Override
public Task register(String type, String action, TransportRequest request) {
    Task task = super.register(type, action, request);
    if (task != null) {
        for (MockTaskManagerListener listener : listeners) {
            try {
                listener.onTaskRegistered(task);
            } catch (Exception e) {
                logger.warn(
                    (Supplier<?>) () -> new ParameterizedMessage(
                        "failed to notify task manager listener about unregistering the task with id {}",
                        task.getId()),
                    e);
            }
        }
    }
    return task;
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的package包/类
/**
 * Registers a task without parent task
 * <p>
 * Returns the task manager tracked task or null if the task doesn't support the task manager
 */
public Task register(String type, String action, TransportRequest request) {
    Task task = request.createTask(taskIdGenerator.incrementAndGet(), type, action, request.getParentTask());
    if (task == null) {
        return null;
    }
    assert task.getParentTaskId().equals(request.getParentTask()) : "Request [ " + request + "] didn't preserve it parentTaskId";
    if (logger.isTraceEnabled()) {
        logger.trace("register {} [{}] [{}] [{}]", task.getId(), type, action, task.getDescription());
    }

    if (task instanceof CancellableTask) {
        registerCancellableTask(task);
    } else {
        Task previousTask = tasks.put(task.getId(), task);
        assert previousTask == null;
    }
    return task;
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
@Override
public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
                                                           ThreadContext threadContext) {
    return Collections.singletonList(new TransportInterceptor() {
        @Override
        public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(String action, String executor,
                                                                                    boolean forceExecution,
                                                                                    TransportRequestHandler<T> actualHandler) {
            return instance.interceptHandler(action, executor, forceExecution, actualHandler);
        }

        @Override
        public AsyncSender interceptSender(AsyncSender sender) {
            return instance.interceptSender(sender);
        }
    });
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的package包/类
@Override
public void messageReceived(T request, TransportChannel channel, Task task) throws Exception {
    synchronized (InterceptingTransportService.this) {
        if (actions.contains(action)) {
            List<TransportRequest> requestList = requests.get(action);
            if (requestList == null) {
                requestList = new ArrayList<>();
                requestList.add(request);
                requests.put(action, requestList);
            } else {
                requestList.add(request);
            }
        }
    }
    requestHandler.messageReceived(request, channel, task);
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的package包/类
@Override
public void logError(Throwable t, TransportRequest request, String action, Task task, int type) {
    switch (type) {
    case 0:
        if(t instanceof ElasticsearchException) {
            auditLog.logMissingPrivileges(action, request, task);
        } else {
            auditLog.logSSLException(request, t, action, task);
        }
        break;
    case 1:
        auditLog.logBadHeaders(request, action, task);
        break;
    default:
        break;
    }
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的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.TransportRequest; //导入依赖的package包/类
@Override
public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(String action, String executor,
                                                                                boolean forceExecution,
                                                                                TransportRequestHandler<T> actualHandler) {
    for (TransportInterceptor interceptor : this.transportInterceptors) {
        actualHandler = interceptor.interceptHandler(action, executor, forceExecution, actualHandler);
    }
    return actualHandler;
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
@Override
protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                           TransportRequestOptions options) throws IOException {
    if (recoveryActionToBlock.equals(action) || requestBlocked.getCount() == 0) {
        logger.info("--> preventing {} request", action);
        requestBlocked.countDown();
        if (dropRequests) {
            return;
        }
        throw new ConnectTransportException(connection.getNode(), "DISCONNECT: prevented " + action + " request");
    }
    super.sendRequest(connection, requestId, action, request, options);
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
@Override
public AsyncSender interceptSender(AsyncSender sender) {
    return new AsyncSender() {
        @Override
        public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action,
                                                              TransportRequest request,
                                                              TransportRequestOptions options,
                                                              TransportResponseHandler<T> handler) {
            final ClusterName clusterName = new ClusterName("cluster1");
            if (TransportLivenessAction.NAME.equals(action)) {
                assertHeaders(threadPool);
                ((TransportResponseHandler<LivenessResponse>) handler).handleResponse(
                    new LivenessResponse(clusterName, connection.getNode()));
            } else if (ClusterStateAction.NAME.equals(action)) {
                assertHeaders(threadPool);
                ClusterName cluster1 = clusterName;
                ClusterState.Builder builder = ClusterState.builder(cluster1);
                //the sniffer detects only data nodes
                builder.nodes(DiscoveryNodes.builder().add(new DiscoveryNode("node_id", "someId", "some_ephemeralId_id",
                    address.address().getHostString(), address.getAddress(), address, Collections.emptyMap(),
                        Collections.singleton(DiscoveryNode.Role.DATA), Version.CURRENT)));
                ((TransportResponseHandler<ClusterStateResponse>) handler)
                        .handleResponse(new ClusterStateResponse(cluster1, builder.build(), 0L));
                clusterStateLatch.countDown();
            } else if (TransportService.HANDSHAKE_ACTION_NAME .equals(action)) {
                ((TransportResponseHandler<TransportService.HandshakeResponse>) handler).handleResponse(
                    new TransportService.HandshakeResponse(connection.getNode(), clusterName, connection.getNode().getVersion()));
            } else {
                handler.handleException(new TransportException("", new InternalException(action)));
            }
        }
    };
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
/**
 * takes a request that was sent by a {@link TransportReplicationAction} and captured
 * and returns the underlying request if it's wrapped or the original (cast to the expected type).
 *
 * This will throw a {@link ClassCastException} if the request is of the wrong type.
 */
public static <R extends ReplicationRequest> R resolveRequest(TransportRequest requestOrWrappedRequest) {
    if (requestOrWrappedRequest instanceof TransportReplicationAction.ConcreteShardRequest) {
        requestOrWrappedRequest = ((TransportReplicationAction.ConcreteShardRequest<?>)requestOrWrappedRequest).getRequest();
    }
    return (R) requestOrWrappedRequest;
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
private void assertConcreteShardRequest(TransportRequest capturedRequest, Request expectedRequest, AllocationId expectedAllocationId) {
    final TransportReplicationAction.ConcreteShardRequest<?> concreteShardRequest =
        (TransportReplicationAction.ConcreteShardRequest<?>) capturedRequest;
    assertThat(concreteShardRequest.getRequest(), equalTo(expectedRequest));
    assertThat(((Request)concreteShardRequest.getRequest()).isRetrySet.get(), equalTo(true));
    assertThat(concreteShardRequest.getTargetAllocationID(), equalTo(expectedAllocationId.getId()));
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
private static void assertSameIndices(IndicesRequest originalRequest, boolean optional, String... actions) {
    for (String action : actions) {
        List<TransportRequest> requests = consumeTransportRequests(action);
        if (!optional) {
            assertThat("no internal requests intercepted for action [" + action + "]", requests.size(), greaterThan(0));
        }
        for (TransportRequest internalRequest : requests) {
            IndicesRequest indicesRequest = convertRequest(internalRequest);
            assertThat(internalRequest.getClass().getName(), indicesRequest.indices(), equalTo(originalRequest.indices()));
            assertThat(indicesRequest.indicesOptions(), equalTo(originalRequest.indicesOptions()));
        }
    }
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
private static void assertIndicesSubset(List<String> indices, String... actions) {
    //indices returned by each bulk shard request need to be a subset of the original indices
    for (String action : actions) {
        List<TransportRequest> requests = consumeTransportRequests(action);
        assertThat("no internal requests intercepted for action [" + action + "]", requests.size(), greaterThan(0));
        for (TransportRequest internalRequest : requests) {
            IndicesRequest indicesRequest = convertRequest(internalRequest);
            for (String index : indicesRequest.indices()) {
                assertThat(indices, hasItem(index));
            }
        }
    }
}
 

import org.elasticsearch.transport.TransportRequest; //导入依赖的package包/类
static IndicesRequest convertRequest(TransportRequest request) {
    final IndicesRequest indicesRequest;
    if (request instanceof IndicesRequest) {
        indicesRequest = (IndicesRequest) request;
    } else {
        indicesRequest = TransportReplicationActionTests.resolveRequest(request);
    }
    return indicesRequest;
}