1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.client;
20
21 import java.io.IOException;
22 import java.net.InetSocketAddress;
23 import java.nio.channels.AsynchronousCloseException;
24 import java.util.ArrayList;
25 import java.util.List;
26 import java.util.Queue;
27 import java.util.concurrent.BlockingQueue;
28 import java.util.concurrent.Future;
29 import java.util.concurrent.RejectedExecutionException;
30 import java.util.concurrent.TimeUnit;
31 import java.util.concurrent.atomic.AtomicInteger;
32
33 import org.eclipse.jetty.client.api.Connection;
34 import org.eclipse.jetty.client.api.Destination;
35 import org.eclipse.jetty.client.api.ProxyConfiguration;
36 import org.eclipse.jetty.client.api.Request;
37 import org.eclipse.jetty.client.api.Response;
38 import org.eclipse.jetty.client.api.Result;
39 import org.eclipse.jetty.http.HttpField;
40 import org.eclipse.jetty.http.HttpHeader;
41 import org.eclipse.jetty.http.HttpMethod;
42 import org.eclipse.jetty.http.HttpScheme;
43 import org.eclipse.jetty.util.BlockingArrayQueue;
44 import org.eclipse.jetty.util.FuturePromise;
45 import org.eclipse.jetty.util.Promise;
46 import org.eclipse.jetty.util.component.ContainerLifeCycle;
47 import org.eclipse.jetty.util.component.Dumpable;
48 import org.eclipse.jetty.util.log.Log;
49 import org.eclipse.jetty.util.log.Logger;
50
51 public class HttpDestination implements Destination, AutoCloseable, Dumpable
52 {
53 private static final Logger LOG = Log.getLogger(HttpDestination.class);
54
55 private final AtomicInteger connectionCount = new AtomicInteger();
56 private final HttpClient client;
57 private final String scheme;
58 private final String host;
59 private final InetSocketAddress address;
60 private final Queue<RequestContext> requests;
61 private final BlockingQueue<Connection> idleConnections;
62 private final BlockingQueue<Connection> activeConnections;
63 private final RequestNotifier requestNotifier;
64 private final ResponseNotifier responseNotifier;
65 private final InetSocketAddress proxyAddress;
66 private final HttpField hostField;
67
68 public HttpDestination(HttpClient client, String scheme, String host, int port)
69 {
70 this.client = client;
71 this.scheme = scheme;
72 this.host = host;
73 this.address = new InetSocketAddress(host, port);
74
75 int maxRequestsQueued = client.getMaxRequestsQueuedPerDestination();
76 int capacity = Math.min(32, maxRequestsQueued);
77 this.requests = new BlockingArrayQueue<>(capacity, capacity, maxRequestsQueued);
78
79 int maxConnections = client.getMaxConnectionsPerDestination();
80 capacity = Math.min(8, maxConnections);
81 this.idleConnections = new BlockingArrayQueue<>(capacity, capacity, maxConnections);
82 this.activeConnections = new BlockingArrayQueue<>(capacity, capacity, maxConnections);
83
84 this.requestNotifier = new RequestNotifier(client);
85 this.responseNotifier = new ResponseNotifier(client);
86
87 ProxyConfiguration proxyConfig = client.getProxyConfiguration();
88 proxyAddress = proxyConfig != null && proxyConfig.matches(host, port) ?
89 new InetSocketAddress(proxyConfig.getHost(), proxyConfig.getPort()) : null;
90
91 String hostValue = host;
92 if ("https".equalsIgnoreCase(scheme) && port != 443 ||
93 "http".equalsIgnoreCase(scheme) && port != 80)
94 hostValue += ":" + port;
95 hostField = new HttpField(HttpHeader.HOST, hostValue);
96 }
97
98 protected BlockingQueue<Connection> getIdleConnections()
99 {
100 return idleConnections;
101 }
102
103 protected BlockingQueue<Connection> getActiveConnections()
104 {
105 return activeConnections;
106 }
107
108 @Override
109 public String getScheme()
110 {
111 return scheme;
112 }
113
114 @Override
115 public String getHost()
116 {
117
118
119 return host;
120 }
121
122 @Override
123 public int getPort()
124 {
125 return address.getPort();
126 }
127
128 public InetSocketAddress getConnectAddress()
129 {
130 return isProxied() ? proxyAddress : address;
131 }
132
133 public boolean isProxied()
134 {
135 return proxyAddress != null;
136 }
137
138 public HttpField getHostField()
139 {
140 return hostField;
141 }
142
143 public void send(Request request, List<Response.ResponseListener> listeners)
144 {
145 if (!scheme.equals(request.getScheme()))
146 throw new IllegalArgumentException("Invalid request scheme " + request.getScheme() + " for destination " + this);
147 if (!getHost().equals(request.getHost()))
148 throw new IllegalArgumentException("Invalid request host " + request.getHost() + " for destination " + this);
149 int port = request.getPort();
150 if (port >= 0 && getPort() != port)
151 throw new IllegalArgumentException("Invalid request port " + port + " for destination " + this);
152
153 RequestContext requestContext = new RequestContext(request, listeners);
154 if (client.isRunning())
155 {
156 if (requests.offer(requestContext))
157 {
158 if (!client.isRunning() && requests.remove(requestContext))
159 {
160 throw new RejectedExecutionException(client + " is stopping");
161 }
162 else
163 {
164 LOG.debug("Queued {}", request);
165 requestNotifier.notifyQueued(request);
166 Connection connection = acquire();
167 if (connection != null)
168 process(connection, false);
169 }
170 }
171 else
172 {
173 throw new RejectedExecutionException("Max requests per destination " + client.getMaxRequestsQueuedPerDestination() + " exceeded");
174 }
175 }
176 else
177 {
178 throw new RejectedExecutionException(client + " is stopped");
179 }
180 }
181
182 public Future<Connection> newConnection()
183 {
184 FuturePromise<Connection> result = new FuturePromise<>();
185 newConnection(new ProxyPromise(result));
186 return result;
187 }
188
189 protected void newConnection(Promise<Connection> promise)
190 {
191 client.newConnection(this, promise);
192 }
193
194 protected Connection acquire()
195 {
196 Connection result = idleConnections.poll();
197 if (result != null)
198 return result;
199
200 final int maxConnections = client.getMaxConnectionsPerDestination();
201 while (true)
202 {
203 int current = connectionCount.get();
204 final int next = current + 1;
205
206 if (next > maxConnections)
207 {
208 LOG.debug("Max connections {} reached for {}", current, this);
209
210 return idleConnections.poll();
211 }
212
213 if (connectionCount.compareAndSet(current, next))
214 {
215 LOG.debug("Creating connection {}/{} for {}", next, maxConnections, this);
216
217
218 Promise<Connection> promise = new Promise<Connection>()
219 {
220 @Override
221 public void succeeded(Connection connection)
222 {
223 process(connection, true);
224 }
225
226 @Override
227 public void failed(final Throwable x)
228 {
229 client.getExecutor().execute(new Runnable()
230 {
231 @Override
232 public void run()
233 {
234 drain(x);
235 }
236 });
237 }
238 };
239
240
241
242
243 newConnection(new ProxyPromise(promise)
244 {
245 @Override
246 public void succeeded(Connection connection)
247 {
248 LOG.debug("Created connection {}/{} {} for {}", next, maxConnections, connection, HttpDestination.this);
249 super.succeeded(connection);
250 }
251
252 @Override
253 public void failed(Throwable x)
254 {
255 LOG.debug("Connection failed {} for {}", x, HttpDestination.this);
256 connectionCount.decrementAndGet();
257 super.failed(x);
258 }
259 });
260
261
262 return idleConnections.poll();
263 }
264 }
265 }
266
267 private void drain(Throwable x)
268 {
269 RequestContext requestContext;
270 while ((requestContext = requests.poll()) != null)
271 {
272 Request request = requestContext.request;
273 requestNotifier.notifyFailure(request, x);
274 List<Response.ResponseListener> listeners = requestContext.listeners;
275 HttpResponse response = new HttpResponse(request, listeners);
276 responseNotifier.notifyFailure(listeners, response, x);
277 responseNotifier.notifyComplete(listeners, new Result(request, x, response, x));
278 }
279 }
280
281
282
283
284
285
286
287
288
289
290 protected void process(Connection connection, boolean dispatch)
291 {
292
293 final HttpConnection httpConnection = (HttpConnection)connection;
294
295 RequestContext requestContext = requests.poll();
296 if (requestContext == null)
297 {
298 LOG.debug("{} idle", httpConnection);
299 if (!idleConnections.offer(httpConnection))
300 {
301 LOG.debug("{} idle overflow");
302 httpConnection.close();
303 }
304 if (!client.isRunning())
305 {
306 LOG.debug("{} is stopping", client);
307 remove(httpConnection);
308 httpConnection.close();
309 }
310 }
311 else
312 {
313 final Request request = requestContext.request;
314 final List<Response.ResponseListener> listeners = requestContext.listeners;
315 Throwable cause = request.getAbortCause();
316 if (cause != null)
317 {
318 abort(request, listeners, cause);
319 LOG.debug("Aborted {} before processing", request);
320 }
321 else
322 {
323 LOG.debug("{} active", httpConnection);
324 if (!activeConnections.offer(httpConnection))
325 {
326 LOG.warn("{} active overflow");
327 }
328 if (dispatch)
329 {
330 client.getExecutor().execute(new Runnable()
331 {
332 @Override
333 public void run()
334 {
335 httpConnection.send(request, listeners);
336 }
337 });
338 }
339 else
340 {
341 httpConnection.send(request, listeners);
342 }
343 }
344 }
345 }
346
347 public void release(Connection connection)
348 {
349 LOG.debug("{} released", connection);
350 if (client.isRunning())
351 {
352 boolean removed = activeConnections.remove(connection);
353 if (removed)
354 process(connection, false);
355 else
356 LOG.debug("{} explicit", connection);
357 }
358 else
359 {
360 LOG.debug("{} is stopped", client);
361 remove(connection);
362 connection.close();
363 }
364 }
365
366 public void remove(Connection connection)
367 {
368 boolean removed = activeConnections.remove(connection);
369 removed |= idleConnections.remove(connection);
370 if (removed)
371 {
372 int open = connectionCount.decrementAndGet();
373 LOG.debug("Removed connection {} for {} - open: {}", connection, this, open);
374 }
375
376
377
378
379 if (!requests.isEmpty())
380 {
381 connection = acquire();
382 if (connection != null)
383 process(connection, false);
384 }
385 }
386
387 public void close()
388 {
389 for (Connection connection : idleConnections)
390 connection.close();
391 idleConnections.clear();
392
393
394 for (Connection connection : activeConnections)
395 connection.close();
396 activeConnections.clear();
397
398 drain(new AsynchronousCloseException());
399
400 connectionCount.set(0);
401
402 LOG.debug("Closed {}", this);
403 }
404
405 public boolean abort(Request request, Throwable cause)
406 {
407 for (RequestContext requestContext : requests)
408 {
409 if (requestContext.request == request)
410 {
411 if (requests.remove(requestContext))
412 {
413
414 abort(request, requestContext.listeners, cause);
415 LOG.debug("Aborted {} while queued", request);
416 return true;
417 }
418 }
419 }
420 return false;
421 }
422
423 private void abort(Request request, List<Response.ResponseListener> listeners, Throwable cause)
424 {
425 HttpResponse response = new HttpResponse(request, listeners);
426 responseNotifier.notifyFailure(listeners, response, cause);
427 responseNotifier.notifyComplete(listeners, new Result(request, cause, response, cause));
428 }
429
430 @Override
431 public String dump()
432 {
433 return ContainerLifeCycle.dump(this);
434 }
435
436 @Override
437 public void dump(Appendable out, String indent) throws IOException
438 {
439 ContainerLifeCycle.dumpObject(out, this + " - requests queued: " + requests.size());
440 List<String> connections = new ArrayList<>();
441 for (Connection connection : idleConnections)
442 connections.add(connection + " - IDLE");
443 for (Connection connection : activeConnections)
444 connections.add(connection + " - ACTIVE");
445 ContainerLifeCycle.dump(out, indent, connections);
446 }
447
448 @Override
449 public String toString()
450 {
451 return String.format("%s(%s://%s:%d)%s",
452 HttpDestination.class.getSimpleName(),
453 getScheme(),
454 getHost(),
455 getPort(),
456 proxyAddress == null ? "" : " via " + proxyAddress.getHostString() + ":" + proxyAddress.getPort());
457 }
458
459 private static class RequestContext
460 {
461 private final Request request;
462 private final List<Response.ResponseListener> listeners;
463
464 private RequestContext(Request request, List<Response.ResponseListener> listeners)
465 {
466 this.request = request;
467 this.listeners = listeners;
468 }
469 }
470
471
472
473
474
475
476
477 private class ProxyPromise implements Promise<Connection>
478 {
479 private final Promise<Connection> delegate;
480
481 private ProxyPromise(Promise<Connection> delegate)
482 {
483 this.delegate = delegate;
484 }
485
486 @Override
487 public void succeeded(Connection connection)
488 {
489 boolean tunnel = isProxied() &&
490 "https".equalsIgnoreCase(getScheme()) &&
491 client.getSslContextFactory() != null;
492 if (tunnel)
493 tunnel(connection);
494 else
495 delegate.succeeded(connection);
496 }
497
498 @Override
499 public void failed(Throwable x)
500 {
501 delegate.failed(x);
502 }
503
504 private void tunnel(final Connection connection)
505 {
506 String target = address.getHostString() + ":" + address.getPort();
507 Request connect = client.newRequest(proxyAddress.getHostString(), proxyAddress.getPort())
508 .scheme(HttpScheme.HTTP.asString())
509 .method(HttpMethod.CONNECT)
510 .path(target)
511 .header(HttpHeader.HOST.asString(), target)
512 .timeout(client.getConnectTimeout(), TimeUnit.MILLISECONDS);
513 connection.send(connect, new Response.CompleteListener()
514 {
515 @Override
516 public void onComplete(Result result)
517 {
518 if (result.isFailed())
519 {
520 failed(result.getFailure());
521 connection.close();
522 }
523 else
524 {
525 Response response = result.getResponse();
526 if (response.getStatus() == 200)
527 {
528 delegate.succeeded(connection);
529 }
530 else
531 {
532 failed(new HttpResponseException("Received " + response + " for " + result.getRequest(), response));
533 connection.close();
534 }
535 }
536 }
537 });
538 }
539 }
540 }