1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
/*
* Licensed to Elasticsearch under one or more contributor
* license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright
* ownership. Elasticsearch licenses this file to you under
* the Apache License, Version 2.0 (the "License"); you may
* not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.client.transport;
import com.carrotsearch.randomizedtesting.generators.RandomInts;
import org.elasticsearch.action.admin.cluster.node.liveness.LivenessResponse;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.component.Lifecycle;
import org.elasticsearch.common.component.LifecycleListener;
import org.elasticsearch.common.transport.BoundTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.transport.ConnectTransportException;
import org.elasticsearch.transport.Transport;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportResponse;
import org.elasticsearch.transport.TransportResponseHandler;
import org.elasticsearch.transport.TransportServiceAdapter;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;
abstract class FailAndRetryMockTransport<Response extends TransportResponse> implements Transport {
private final Random random;
private boolean connectMode = true;
private TransportServiceAdapter transportServiceAdapter;
private final AtomicInteger connectTransportExceptions = new AtomicInteger();
private final AtomicInteger failures = new AtomicInteger();
private final AtomicInteger successes = new AtomicInteger();
private final Set<DiscoveryNode> triedNodes = new CopyOnWriteArraySet<>();
FailAndRetryMockTransport(Random random) {
this.random = new Random(random.nextLong());
}
@Override
@SuppressWarnings("unchecked")
public void sendRequest(DiscoveryNode node, long requestId, String action, TransportRequest request, TransportRequestOptions options) throws IOException, TransportException {
//we make sure that nodes get added to the connected ones when calling addTransportAddress, by returning proper nodes info
if (connectMode) {
TransportResponseHandler transportResponseHandler = transportServiceAdapter.onResponseReceived(requestId);
transportResponseHandler.handleResponse(new LivenessResponse(ClusterName.DEFAULT, node));
return;
}
//once nodes are connected we'll just return errors for each sendRequest call
triedNodes.add(node);
if (RandomInts.randomInt(random, 100) > 10) {
connectTransportExceptions.incrementAndGet();
throw new ConnectTransportException(node, "node not available");
} else {
if (random.nextBoolean()) {
failures.incrementAndGet();
//throw whatever exception that is not a subclass of ConnectTransportException
throw new IllegalStateException();
} else {
TransportResponseHandler transportResponseHandler = transportServiceAdapter.onResponseReceived(requestId);
if (random.nextBoolean()) {
successes.incrementAndGet();
transportResponseHandler.handleResponse(newResponse());
} else {
failures.incrementAndGet();
transportResponseHandler.handleException(new TransportException("transport exception"));
}
}
}
}
protected abstract Response newResponse();
public void endConnectMode() {
this.connectMode = false;
}
public int connectTransportExceptions() {
return connectTransportExceptions.get();
}
public int failures() {
return failures.get();
}
public int successes() {
return successes.get();
}
public Set<DiscoveryNode> triedNodes() {
return triedNodes;
}
@Override
public void transportServiceAdapter(TransportServiceAdapter transportServiceAdapter) {
this.transportServiceAdapter = transportServiceAdapter;
}
@Override
public BoundTransportAddress boundAddress() {
return null;
}
@Override
public TransportAddress[] addressesFromString(String address, int perAddressLimit) throws Exception {
throw new UnsupportedOperationException();
}
@Override
public boolean addressSupported(Class<? extends TransportAddress> address) {
throw new UnsupportedOperationException();
}
@Override
public boolean nodeConnected(DiscoveryNode node) {
return false;
}
@Override
public void connectToNode(DiscoveryNode node) throws ConnectTransportException {
}
@Override
public void connectToNodeLight(DiscoveryNode node) throws ConnectTransportException {
}
@Override
public void disconnectFromNode(DiscoveryNode node) {
}
@Override
public long serverOpen() {
return 0;
}
@Override
public Lifecycle.State lifecycleState() {
return null;
}
@Override
public void addLifecycleListener(LifecycleListener listener) {
throw new UnsupportedOperationException();
}
@Override
public void removeLifecycleListener(LifecycleListener listener) {
throw new UnsupportedOperationException();
}
@Override
public Transport start() {
return null;
}
@Override
public Transport stop() {
return null;
}
@Override
public void close() {
}
@Override
public Map<String, BoundTransportAddress> profileBoundAddresses() {
return Collections.emptyMap();
}
}
|
