Allow extension of CapturingTransport by subclasses (#33012)

Today, CapturingTransport#createCapturingTransportService creates a transport
service with a connection manager with reasonable default behaviours, but
overriding this behaviour in a consumer is a litle tricky. Additionally, the
default behaviour for opening a connection duplicates the content of the
CapturingTransport#openConnection() method.

This change removes this duplication by delegating to openConnection() and
introduces overridable nodeConnected() and onSendRequest() methods so that
consumers can alter this behaviour more easily.

Relates #32246 in which we test the mechanisms for opening connections to
unknown (and possibly unreachable) nodes.
This commit is contained in:
David Turner 2018-08-22 09:09:08 +01:00 committed by GitHub
parent ffb1a5d5b7
commit ab000323fa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 37 additions and 50 deletions

View File

@ -51,7 +51,6 @@ import org.elasticsearch.transport.TransportService;
import org.elasticsearch.transport.TransportStats; import org.elasticsearch.transport.TransportStats;
import java.io.IOException; import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.Collections; import java.util.Collections;
@ -66,11 +65,13 @@ import java.util.function.Function;
import static org.apache.lucene.util.LuceneTestCase.rarely; import static org.apache.lucene.util.LuceneTestCase.rarely;
/** A transport class that doesn't send anything but rather captures all requests for inspection from tests */ /**
* A transport class that doesn't send anything but rather captures all requests for inspection from tests
*/
public class CapturingTransport implements Transport { public class CapturingTransport implements Transport {
private volatile Map<String, RequestHandlerRegistry> requestHandlers = Collections.emptyMap(); private volatile Map<String, RequestHandlerRegistry> requestHandlers = Collections.emptyMap();
final Object requestHandlerMutex = new Object(); private final Object requestHandlerMutex = new Object();
private final ResponseHandlers responseHandlers = new ResponseHandlers(); private final ResponseHandlers responseHandlers = new ResponseHandlers();
private TransportMessageListener listener; private TransportMessageListener listener;
@ -80,7 +81,7 @@ public class CapturingTransport implements Transport {
public final String action; public final String action;
public final TransportRequest request; public final TransportRequest request;
public CapturedRequest(DiscoveryNode node, long requestId, String action, TransportRequest request) { CapturedRequest(DiscoveryNode node, long requestId, String action, TransportRequest request) {
this.node = node; this.node = node;
this.requestId = requestId; this.requestId = requestId;
this.action = action; this.action = action;
@ -96,41 +97,15 @@ public class CapturingTransport implements Transport {
@Nullable ClusterSettings clusterSettings, Set<String> taskHeaders) { @Nullable ClusterSettings clusterSettings, Set<String> taskHeaders) {
StubbableConnectionManager connectionManager = new StubbableConnectionManager(new ConnectionManager(settings, this, threadPool), StubbableConnectionManager connectionManager = new StubbableConnectionManager(new ConnectionManager(settings, this, threadPool),
settings, this, threadPool); settings, this, threadPool);
connectionManager.setDefaultNodeConnectedBehavior((cm, discoveryNode) -> true); connectionManager.setDefaultNodeConnectedBehavior((cm, discoveryNode) -> nodeConnected(discoveryNode));
connectionManager.setDefaultConnectBehavior((cm, discoveryNode) -> new Connection() { connectionManager.setDefaultConnectBehavior((cm, discoveryNode) -> openConnection(discoveryNode, null));
@Override
public DiscoveryNode getNode() {
return discoveryNode;
}
@Override
public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
throws TransportException {
requests.put(requestId, Tuple.tuple(discoveryNode, action));
capturedRequests.add(new CapturedRequest(discoveryNode, requestId, action, request));
}
@Override
public void addCloseListener(ActionListener<Void> listener) {
}
@Override
public boolean isClosed() {
return false;
}
@Override
public void close() {
}
});
return new TransportService(settings, this, threadPool, interceptor, localNodeFactory, clusterSettings, taskHeaders, return new TransportService(settings, this, threadPool, interceptor, localNodeFactory, clusterSettings, taskHeaders,
connectionManager); connectionManager);
} }
/** returns all requests captured so far. Doesn't clear the captured request list. See {@link #clear()} */ /**
* returns all requests captured so far. Doesn't clear the captured request list. See {@link #clear()}
*/
public CapturedRequest[] capturedRequests() { public CapturedRequest[] capturedRequests() {
return capturedRequests.toArray(new CapturedRequest[0]); return capturedRequests.toArray(new CapturedRequest[0]);
} }
@ -178,12 +153,16 @@ public class CapturingTransport implements Transport {
return groupRequestsByTargetNode(requests); return groupRequestsByTargetNode(requests);
} }
/** clears captured requests */ /**
* clears captured requests
*/
public void clear() { public void clear() {
capturedRequests.clear(); capturedRequests.clear();
} }
/** simulate a response for the given requestId */ /**
* simulate a response for the given requestId
*/
public void handleResponse(final long requestId, final TransportResponse response) { public void handleResponse(final long requestId, final TransportResponse response) {
responseHandlers.onResponseReceived(requestId, listener).handleResponse(response); responseHandlers.onResponseReceived(requestId, listener).handleResponse(response);
} }
@ -194,7 +173,7 @@ public class CapturingTransport implements Transport {
* *
* @param requestId the id corresponding to the captured send * @param requestId the id corresponding to the captured send
* request * request
* @param t the failure to wrap * @param t the failure to wrap
*/ */
public void handleLocalError(final long requestId, final Throwable t) { public void handleLocalError(final long requestId, final Throwable t) {
Tuple<DiscoveryNode, String> request = requests.get(requestId); Tuple<DiscoveryNode, String> request = requests.get(requestId);
@ -208,7 +187,7 @@ public class CapturingTransport implements Transport {
* *
* @param requestId the id corresponding to the captured send * @param requestId the id corresponding to the captured send
* request * request
* @param t the failure to wrap * @param t the failure to wrap
*/ */
public void handleRemoteError(final long requestId, final Throwable t) { public void handleRemoteError(final long requestId, final Throwable t) {
final RemoteTransportException remoteException; final RemoteTransportException remoteException;
@ -234,7 +213,7 @@ public class CapturingTransport implements Transport {
* *
* @param requestId the id corresponding to the captured send * @param requestId the id corresponding to the captured send
* request * request
* @param e the failure * @param e the failure
*/ */
public void handleError(final long requestId, final TransportException e) { public void handleError(final long requestId, final TransportException e) {
responseHandlers.onResponseReceived(requestId, listener).handleException(e); responseHandlers.onResponseReceived(requestId, listener).handleException(e);
@ -251,13 +230,11 @@ public class CapturingTransport implements Transport {
@Override @Override
public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options) public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
throws TransportException { throws TransportException {
requests.put(requestId, Tuple.tuple(node, action)); onSendRequest(requestId, action, request, node);
capturedRequests.add(new CapturedRequest(node, requestId, action, request));
} }
@Override @Override
public void addCloseListener(ActionListener<Void> listener) { public void addCloseListener(ActionListener<Void> listener) {
} }
@Override @Override
@ -267,11 +244,19 @@ public class CapturingTransport implements Transport {
@Override @Override
public void close() { public void close() {
} }
}; };
} }
protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode node) {
requests.put(requestId, Tuple.tuple(node, action));
capturedRequests.add(new CapturedRequest(node, requestId, action, request));
}
protected boolean nodeConnected(DiscoveryNode discoveryNode) {
return true;
}
@Override @Override
public TransportStats getStats() { public TransportStats getStats() {
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
@ -288,7 +273,7 @@ public class CapturingTransport implements Transport {
} }
@Override @Override
public TransportAddress[] addressesFromString(String address, int perAddressLimit) throws UnknownHostException { public TransportAddress[] addressesFromString(String address, int perAddressLimit) {
return new TransportAddress[0]; return new TransportAddress[0];
} }
@ -299,22 +284,23 @@ public class CapturingTransport implements Transport {
@Override @Override
public void addLifecycleListener(LifecycleListener listener) { public void addLifecycleListener(LifecycleListener listener) {
} }
@Override @Override
public void removeLifecycleListener(LifecycleListener listener) { public void removeLifecycleListener(LifecycleListener listener) {
} }
@Override @Override
public void start() {} public void start() {
}
@Override @Override
public void stop() {} public void stop() {
}
@Override @Override
public void close() {} public void close() {
}
@Override @Override
public List<String> getLocalAddresses() { public List<String> getLocalAddresses() {
@ -330,6 +316,7 @@ public class CapturingTransport implements Transport {
requestHandlers = MapBuilder.newMapBuilder(requestHandlers).put(reg.getAction(), reg).immutableMap(); requestHandlers = MapBuilder.newMapBuilder(requestHandlers).put(reg.getAction(), reg).immutableMap();
} }
} }
@Override @Override
public ResponseHandlers getResponseHandlers() { public ResponseHandlers getResponseHandlers() {
return responseHandlers; return responseHandlers;