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