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 package org.apache.hc.client5.http.impl.classic;
29
30 import java.io.IOException;
31 import java.util.concurrent.ExecutionException;
32 import java.util.concurrent.TimeoutException;
33 import java.util.concurrent.atomic.AtomicReference;
34
35 import org.apache.hc.client5.http.HttpRoute;
36 import org.apache.hc.client5.http.classic.ExecRuntime;
37 import org.apache.hc.client5.http.config.RequestConfig;
38 import org.apache.hc.client5.http.impl.ConnPoolSupport;
39 import org.apache.hc.client5.http.io.ConnectionEndpoint;
40 import org.apache.hc.client5.http.io.HttpClientConnectionManager;
41 import org.apache.hc.client5.http.io.LeaseRequest;
42 import org.apache.hc.client5.http.protocol.HttpClientContext;
43 import org.apache.hc.core5.concurrent.Cancellable;
44 import org.apache.hc.core5.concurrent.CancellableDependency;
45 import org.apache.hc.core5.http.ClassicHttpRequest;
46 import org.apache.hc.core5.http.ClassicHttpResponse;
47 import org.apache.hc.core5.http.ConnectionRequestTimeoutException;
48 import org.apache.hc.core5.http.HttpException;
49 import org.apache.hc.core5.http.impl.io.HttpRequestExecutor;
50 import org.apache.hc.core5.io.CloseMode;
51 import org.apache.hc.core5.util.Args;
52 import org.apache.hc.core5.util.TimeValue;
53 import org.apache.hc.core5.util.Timeout;
54 import org.slf4j.Logger;
55
56 class InternalExecRuntime implements ExecRuntime, Cancellable {
57
58 private final Logger log;
59
60 private final HttpClientConnectionManager manager;
61 private final HttpRequestExecutor requestExecutor;
62 private final CancellableDependency cancellableDependency;
63 private final AtomicReference<ConnectionEndpoint> endpointRef;
64
65 private volatile boolean reusable;
66 private volatile Object state;
67 private volatile TimeValue validDuration;
68
69 InternalExecRuntime(
70 final Logger log,
71 final HttpClientConnectionManager manager,
72 final HttpRequestExecutor requestExecutor,
73 final CancellableDependency cancellableDependency) {
74 super();
75 this.log = log;
76 this.manager = manager;
77 this.requestExecutor = requestExecutor;
78 this.cancellableDependency = cancellableDependency;
79 this.endpointRef = new AtomicReference<>();
80 this.validDuration = TimeValue.NEG_ONE_MILLISECOND;
81 }
82
83 @Override
84 public boolean isExecutionAborted() {
85 return cancellableDependency != null && cancellableDependency.isCancelled();
86 }
87
88 @Override
89 public boolean isEndpointAcquired() {
90 return endpointRef.get() != null;
91 }
92
93 @Override
94 public void acquireEndpoint(
95 final String id, final HttpRoute route, final Object object, final HttpClientContext context) throws IOException {
96 Args.notNull(route, "Route");
97 if (endpointRef.get() == null) {
98 final RequestConfig requestConfig = context.getRequestConfig();
99 final Timeout connectionRequestTimeout = requestConfig.getConnectionRequestTimeout();
100 if (log.isDebugEnabled()) {
101 log.debug("{} acquiring endpoint ({})", id, connectionRequestTimeout);
102 }
103 final LeaseRequest connRequest = manager.lease(id, route, connectionRequestTimeout, object);
104 state = object;
105 if (cancellableDependency != null) {
106 cancellableDependency.setDependency(connRequest);
107 }
108 try {
109 final ConnectionEndpoint connectionEndpoint = connRequest.get(connectionRequestTimeout);
110 endpointRef.set(connectionEndpoint);
111 reusable = connectionEndpoint.isConnected();
112 if (cancellableDependency != null) {
113 cancellableDependency.setDependency(this);
114 }
115 if (log.isDebugEnabled()) {
116 log.debug("{} acquired endpoint {}", id, ConnPoolSupport.getId(connectionEndpoint));
117 }
118 } catch(final TimeoutException ex) {
119 connRequest.cancel();
120 throw new ConnectionRequestTimeoutException(ex.getMessage());
121 } catch(final InterruptedException interrupted) {
122 connRequest.cancel();
123 Thread.currentThread().interrupt();
124 throw new RequestFailedException("Request aborted", interrupted);
125 } catch(final ExecutionException ex) {
126 connRequest.cancel();
127 Throwable cause = ex.getCause();
128 if (cause == null) {
129 cause = ex;
130 }
131 throw new RequestFailedException("Request execution failed", cause);
132 }
133 } else {
134 throw new IllegalStateException("Endpoint already acquired");
135 }
136 }
137
138 ConnectionEndpoint ensureValid() {
139 final ConnectionEndpoint endpoint = endpointRef.get();
140 if (endpoint == null) {
141 throw new IllegalStateException("Endpoint not acquired / already released");
142 }
143 return endpoint;
144 }
145
146 @Override
147 public boolean isEndpointConnected() {
148 final ConnectionEndpoint endpoint = endpointRef.get();
149 return endpoint != null && endpoint.isConnected();
150 }
151
152 private void connectEndpoint(final ConnectionEndpoint endpoint, final HttpClientContext context) throws IOException {
153 if (isExecutionAborted()) {
154 throw new RequestFailedException("Request aborted");
155 }
156 final RequestConfig requestConfig = context.getRequestConfig();
157 @SuppressWarnings("deprecation")
158 final Timeout connectTimeout = requestConfig.getConnectTimeout();
159 if (log.isDebugEnabled()) {
160 log.debug("{} connecting endpoint ({})", ConnPoolSupport.getId(endpoint), connectTimeout);
161 }
162 manager.connect(endpoint, connectTimeout, context);
163 if (log.isDebugEnabled()) {
164 log.debug("{} endpoint connected", ConnPoolSupport.getId(endpoint));
165 }
166 }
167
168 @Override
169 public void connectEndpoint(final HttpClientContext context) throws IOException {
170 final ConnectionEndpoint endpoint = ensureValid();
171 if (!endpoint.isConnected()) {
172 connectEndpoint(endpoint, context);
173 }
174 }
175
176 @Override
177 public void disconnectEndpoint() throws IOException {
178 final ConnectionEndpoint endpoint = endpointRef.get();
179 if (endpoint != null) {
180 endpoint.close();
181 if (log.isDebugEnabled()) {
182 log.debug("{} endpoint closed", ConnPoolSupport.getId(endpoint));
183 }
184 }
185 }
186
187 @Override
188 public void upgradeTls(final HttpClientContext context) throws IOException {
189 final ConnectionEndpoint endpoint = ensureValid();
190 if (log.isDebugEnabled()) {
191 log.debug("{} upgrading endpoint", ConnPoolSupport.getId(endpoint));
192 }
193 manager.upgrade(endpoint, context);
194 }
195
196 @Override
197 public ClassicHttpResponse execute(
198 final String id,
199 final ClassicHttpRequest request,
200 final HttpClientContext context) throws IOException, HttpException {
201 final ConnectionEndpoint endpoint = ensureValid();
202 if (!endpoint.isConnected()) {
203 connectEndpoint(endpoint, context);
204 }
205 if (isExecutionAborted()) {
206 throw new RequestFailedException("Request aborted");
207 }
208 final RequestConfig requestConfig = context.getRequestConfig();
209 final Timeout responseTimeout = requestConfig.getResponseTimeout();
210 if (responseTimeout != null) {
211 endpoint.setSocketTimeout(responseTimeout);
212 }
213 if (log.isDebugEnabled()) {
214 log.debug("{} start execution {}", ConnPoolSupport.getId(endpoint), id);
215 }
216 return endpoint.execute(id, request, requestExecutor, context);
217 }
218
219 @Override
220 public boolean isConnectionReusable() {
221 return reusable;
222 }
223
224 @Override
225 public void markConnectionReusable(final Object state, final TimeValue validDuration) {
226 this.reusable = true;
227 this.state = state;
228 this.validDuration = validDuration;
229 }
230
231 @Override
232 public void markConnectionNonReusable() {
233 reusable = false;
234 }
235
236 private void discardEndpoint(final ConnectionEndpoint endpoint) {
237 try {
238 endpoint.close(CloseMode.IMMEDIATE);
239 if (log.isDebugEnabled()) {
240 log.debug("{} endpoint closed", ConnPoolSupport.getId(endpoint));
241 }
242 } finally {
243 if (log.isDebugEnabled()) {
244 log.debug("{} discarding endpoint", ConnPoolSupport.getId(endpoint));
245 }
246 manager.release(endpoint, null, TimeValue.ZERO_MILLISECONDS);
247 }
248 }
249
250 @Override
251 public void releaseEndpoint() {
252 final ConnectionEndpoint endpoint = endpointRef.getAndSet(null);
253 if (endpoint != null) {
254 if (reusable) {
255 if (log.isDebugEnabled()) {
256 log.debug("{} releasing valid endpoint", ConnPoolSupport.getId(endpoint));
257 }
258 manager.release(endpoint, state, validDuration);
259 } else {
260 discardEndpoint(endpoint);
261 }
262 }
263 }
264
265 @Override
266 public void discardEndpoint() {
267 final ConnectionEndpoint endpoint = endpointRef.getAndSet(null);
268 if (endpoint != null) {
269 discardEndpoint(endpoint);
270 }
271 }
272
273 @Override
274 public boolean cancel() {
275 final boolean alreadyReleased = endpointRef.get() == null;
276 final ConnectionEndpoint endpoint = endpointRef.getAndSet(null);
277 if (endpoint != null) {
278 if (log.isDebugEnabled()) {
279 log.debug("{} cancel", ConnPoolSupport.getId(endpoint));
280 }
281 discardEndpoint(endpoint);
282 }
283 return !alreadyReleased;
284 }
285
286 @Override
287 public ExecRuntime fork(final CancellableDependency cancellableDependency) {
288 return new InternalExecRuntime(log, manager, requestExecutor, cancellableDependency);
289 }
290
291 }