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.lang.reflect.Constructor;
23 import java.net.ProtocolException;
24 import java.util.ArrayList;
25 import java.util.LinkedList;
26 import java.util.List;
27 import java.util.concurrent.ArrayBlockingQueue;
28 import java.util.concurrent.BlockingQueue;
29 import java.util.concurrent.RejectedExecutionException;
30
31 import org.eclipse.jetty.client.HttpClient.Connector;
32 import org.eclipse.jetty.client.security.Authentication;
33 import org.eclipse.jetty.client.security.SecurityListener;
34 import org.eclipse.jetty.http.HttpCookie;
35 import org.eclipse.jetty.http.HttpHeaders;
36 import org.eclipse.jetty.http.HttpMethods;
37 import org.eclipse.jetty.http.HttpStatus;
38 import org.eclipse.jetty.http.PathMap;
39 import org.eclipse.jetty.io.Buffer;
40 import org.eclipse.jetty.io.ByteArrayBuffer;
41 import org.eclipse.jetty.io.Connection;
42 import org.eclipse.jetty.io.EndPoint;
43 import org.eclipse.jetty.util.component.AggregateLifeCycle;
44 import org.eclipse.jetty.util.component.Dumpable;
45 import org.eclipse.jetty.util.log.Log;
46 import org.eclipse.jetty.util.log.Logger;
47 import org.eclipse.jetty.util.ssl.SslContextFactory;
48
49
50
51
52 public class HttpDestination implements Dumpable
53 {
54 private static final Logger LOG = Log.getLogger(HttpDestination.class);
55
56 private final List<HttpExchange> _exchanges = new LinkedList<HttpExchange>();
57 private final List<AbstractHttpConnection> _connections = new LinkedList<AbstractHttpConnection>();
58 private final BlockingQueue<Object> _reservedConnections = new ArrayBlockingQueue<Object>(10, true);
59 private final List<AbstractHttpConnection> _idleConnections = new ArrayList<AbstractHttpConnection>();
60 private final HttpClient _client;
61 private final Address _address;
62 private final boolean _ssl;
63 private final SslContextFactory _sslContextFactory;
64 private final ByteArrayBuffer _hostHeader;
65 private volatile int _maxConnections;
66 private volatile int _maxQueueSize;
67 private int _pendingConnections = 0;
68 private int _pendingReservedConnections = 0;
69 private volatile Address _proxy;
70 private Authentication _proxyAuthentication;
71 private PathMap _authorizations;
72 private List<HttpCookie> _cookies;
73
74 HttpDestination(HttpClient client, Address address, boolean ssl, SslContextFactory sslContextFactory)
75 {
76 _client = client;
77 _address = address;
78 _ssl = ssl;
79 _sslContextFactory = sslContextFactory;
80 _maxConnections = _client.getMaxConnectionsPerAddress();
81 _maxQueueSize = _client.getMaxQueueSizePerAddress();
82 String addressString = address.getHost();
83 if (address.getPort() != (_ssl ? 443 : 80))
84 addressString += ":" + address.getPort();
85 _hostHeader = new ByteArrayBuffer(addressString);
86 }
87
88 public HttpClient getHttpClient()
89 {
90 return _client;
91 }
92
93 public Address getAddress()
94 {
95 return _address;
96 }
97
98 public boolean isSecure()
99 {
100 return _ssl;
101 }
102
103 public SslContextFactory getSslContextFactory()
104 {
105 return _sslContextFactory;
106 }
107
108 public Buffer getHostHeader()
109 {
110 return _hostHeader;
111 }
112
113 public int getMaxConnections()
114 {
115 return _maxConnections;
116 }
117
118 public void setMaxConnections(int maxConnections)
119 {
120 this._maxConnections = maxConnections;
121 }
122
123 public int getMaxQueueSize()
124 {
125 return _maxQueueSize;
126 }
127
128 public void setMaxQueueSize(int maxQueueSize)
129 {
130 this._maxQueueSize = maxQueueSize;
131 }
132
133 public int getConnections()
134 {
135 synchronized (this)
136 {
137 return _connections.size();
138 }
139 }
140
141 public int getIdleConnections()
142 {
143 synchronized (this)
144 {
145 return _idleConnections.size();
146 }
147 }
148
149 public void addAuthorization(String pathSpec, Authentication authorization)
150 {
151 synchronized (this)
152 {
153 if (_authorizations == null)
154 _authorizations = new PathMap();
155 _authorizations.put(pathSpec, authorization);
156 }
157
158
159 }
160
161 public void addCookie(HttpCookie cookie)
162 {
163 synchronized (this)
164 {
165 if (_cookies == null)
166 _cookies = new ArrayList<HttpCookie>();
167 _cookies.add(cookie);
168 }
169
170
171 }
172
173 public void clearCookies()
174 {
175 synchronized (this)
176 {
177 _cookies.clear();
178 }
179
180 }
181
182
183
184
185
186
187
188
189
190
191 private AbstractHttpConnection getConnection(long timeout) throws IOException
192 {
193 AbstractHttpConnection connection = null;
194
195 while ((connection == null) && (connection = getIdleConnection()) == null && timeout > 0)
196 {
197 boolean startConnection = false;
198 synchronized (this)
199 {
200 int totalConnections = _connections.size() + _pendingConnections;
201 if (totalConnections < _maxConnections)
202 {
203 _pendingReservedConnections++;
204 startConnection = true;
205 }
206 }
207
208 if (startConnection)
209 {
210 startNewConnection();
211 try
212 {
213 Object o = _reservedConnections.take();
214 if (o instanceof AbstractHttpConnection)
215 {
216 connection = (AbstractHttpConnection)o;
217 }
218 else
219 throw (IOException)o;
220 }
221 catch (InterruptedException e)
222 {
223 LOG.ignore(e);
224 }
225 }
226 else
227 {
228 try
229 {
230 Thread.currentThread();
231 Thread.sleep(200);
232 timeout -= 200;
233 }
234 catch (InterruptedException e)
235 {
236 LOG.ignore(e);
237 }
238 }
239 }
240 return connection;
241 }
242
243 public AbstractHttpConnection reserveConnection(long timeout) throws IOException
244 {
245 AbstractHttpConnection connection = getConnection(timeout);
246 if (connection != null)
247 connection.setReserved(true);
248 return connection;
249 }
250
251 public AbstractHttpConnection getIdleConnection() throws IOException
252 {
253 AbstractHttpConnection connection = null;
254 while (true)
255 {
256 synchronized (this)
257 {
258 if (connection != null)
259 {
260 _connections.remove(connection);
261 connection.close();
262 connection = null;
263 }
264 if (_idleConnections.size() > 0)
265 connection = _idleConnections.remove(_idleConnections.size() - 1);
266 }
267
268 if (connection == null)
269 {
270 return null;
271 }
272
273
274
275 if (connection.cancelIdleTimeout())
276 {
277 return connection;
278 }
279 }
280 }
281
282 protected void startNewConnection()
283 {
284 try
285 {
286 synchronized (this)
287 {
288 _pendingConnections++;
289 }
290 final Connector connector = _client._connector;
291 if (connector != null)
292 connector.startConnection(this);
293 }
294 catch (Exception e)
295 {
296 LOG.debug(e);
297 onConnectionFailed(e);
298 }
299 }
300
301 public void onConnectionFailed(Throwable throwable)
302 {
303 Throwable connect_failure = null;
304
305 boolean startConnection = false;
306 synchronized (this)
307 {
308 _pendingConnections--;
309 if (_pendingReservedConnections > 0)
310 {
311 connect_failure = throwable;
312 _pendingReservedConnections--;
313 }
314 else if (_exchanges.size() > 0)
315 {
316 HttpExchange ex = _exchanges.remove(0);
317 if (ex.setStatus(HttpExchange.STATUS_EXCEPTED))
318 ex.getEventListener().onConnectionFailed(throwable);
319
320
321
322 if (!_exchanges.isEmpty() && _client.isStarted())
323 startConnection = true;
324 }
325 }
326
327 if (startConnection)
328 startNewConnection();
329
330 if (connect_failure != null)
331 {
332 try
333 {
334 _reservedConnections.put(connect_failure);
335 }
336 catch (InterruptedException e)
337 {
338 LOG.ignore(e);
339 }
340 }
341 }
342
343 public void onException(Throwable throwable)
344 {
345 synchronized (this)
346 {
347 _pendingConnections--;
348 if (_exchanges.size() > 0)
349 {
350 HttpExchange ex = _exchanges.remove(0);
351 if (ex.setStatus(HttpExchange.STATUS_EXCEPTED))
352 ex.getEventListener().onException(throwable);
353 }
354 }
355 }
356
357 public void onNewConnection(final AbstractHttpConnection connection) throws IOException
358 {
359 Connection reservedConnection = null;
360
361 synchronized (this)
362 {
363 _pendingConnections--;
364 _connections.add(connection);
365
366 if (_pendingReservedConnections > 0)
367 {
368 reservedConnection = connection;
369 _pendingReservedConnections--;
370 }
371 else
372 {
373
374 EndPoint endPoint = connection.getEndPoint();
375 if (isProxied() && endPoint instanceof SelectConnector.UpgradableEndPoint)
376 {
377 SelectConnector.UpgradableEndPoint proxyEndPoint = (SelectConnector.UpgradableEndPoint)endPoint;
378 ConnectExchange connect = new ConnectExchange(getAddress(), proxyEndPoint);
379 connect.setAddress(getProxy());
380 LOG.debug("Establishing tunnel to {} via {}", getAddress(), getProxy());
381 send(connection, connect);
382 }
383 else
384 {
385
386 if (_exchanges.size() == 0)
387 {
388 LOG.debug("No exchanges for new connection {}", connection);
389 connection.setIdleTimeout();
390 _idleConnections.add(connection);
391 }
392 else
393 {
394 HttpExchange exchange = _exchanges.remove(0);
395 send(connection, exchange);
396 }
397 }
398 }
399 }
400
401 if (reservedConnection != null)
402 {
403 try
404 {
405 _reservedConnections.put(reservedConnection);
406 }
407 catch (InterruptedException e)
408 {
409 LOG.ignore(e);
410 }
411 }
412 }
413
414 public void returnConnection(AbstractHttpConnection connection, boolean close) throws IOException
415 {
416 if (connection.isReserved())
417 connection.setReserved(false);
418
419 if (close)
420 {
421 try
422 {
423 connection.close();
424 }
425 catch (IOException e)
426 {
427 LOG.ignore(e);
428 }
429 }
430
431 if (!_client.isStarted())
432 return;
433
434 if (!close && connection.getEndPoint().isOpen())
435 {
436 synchronized (this)
437 {
438 if (_exchanges.size() == 0)
439 {
440 connection.setIdleTimeout();
441 _idleConnections.add(connection);
442 }
443 else
444 {
445 HttpExchange ex = _exchanges.remove(0);
446 send(connection, ex);
447 }
448 this.notifyAll();
449 }
450 }
451 else
452 {
453 boolean startConnection = false;
454 boolean remove = false;
455 synchronized (this)
456 {
457 _connections.remove(connection);
458 if (_exchanges.isEmpty())
459 remove=_client.isRemoveIdleDestinations() && ( _cookies==null || _cookies.isEmpty() ) &&_connections.isEmpty() && _idleConnections.isEmpty();
460 else if (_client.isStarted())
461 startConnection = true;
462 }
463
464 if (startConnection)
465 startNewConnection();
466 if (remove)
467 _client.removeDestination(this);
468 }
469 }
470
471 public void returnIdleConnection(AbstractHttpConnection connection)
472 {
473
474 long idleForMs = connection.getEndPoint() != null ? connection.getEndPoint().getMaxIdleTime() : -1;
475 connection.onIdleExpired(idleForMs);
476
477 boolean startConnection = false;
478 boolean remove = false;
479 synchronized (this)
480 {
481 _idleConnections.remove(connection);
482 _connections.remove(connection);
483
484 if (_exchanges.isEmpty())
485 remove=_client.isRemoveIdleDestinations() && ( _cookies==null || _cookies.isEmpty() ) &&_connections.isEmpty() && _idleConnections.isEmpty();
486 else if (_client.isStarted())
487 startConnection = true;
488 }
489
490 if (startConnection)
491 startNewConnection();
492 if (remove)
493 _client.removeDestination(this);
494 }
495
496 public void send(HttpExchange ex) throws IOException
497 {
498 ex.setStatus(HttpExchange.STATUS_WAITING_FOR_CONNECTION);
499
500 LinkedList<String> listeners = _client.getRegisteredListeners();
501 if (listeners != null)
502 {
503
504 for (int i = listeners.size(); i > 0; --i)
505 {
506 String listenerClass = listeners.get(i - 1);
507 try
508 {
509 Class<?> listener = Class.forName(listenerClass);
510 Constructor constructor = listener.getDeclaredConstructor(HttpDestination.class, HttpExchange.class);
511 HttpEventListener elistener = (HttpEventListener)constructor.newInstance(this, ex);
512 ex.setEventListener(elistener);
513 }
514 catch (final Exception e)
515 {
516 throw new IOException("Unable to instantiate registered listener for destination: " + listenerClass)
517 {
518 {
519 initCause(e);
520 }
521 };
522 }
523 }
524 }
525
526
527 if (_client.hasRealms())
528 {
529 ex.setEventListener(new SecurityListener(this, ex));
530 }
531
532 doSend(ex);
533 }
534
535 public void resend(HttpExchange ex) throws IOException
536 {
537 ex.getEventListener().onRetry();
538 ex.reset();
539 doSend(ex);
540 }
541
542 protected void doSend(HttpExchange ex) throws IOException
543 {
544
545
546 synchronized (this)
547 {
548 if (_cookies != null)
549 {
550 StringBuilder buf = null;
551 for (HttpCookie cookie : _cookies)
552 {
553 if (buf == null)
554 buf = new StringBuilder();
555 else
556 buf.append("; ");
557 buf.append(cookie.getName());
558 buf.append("=");
559 buf.append(cookie.getValue());
560 }
561 if (buf != null)
562 ex.addRequestHeader(HttpHeaders.COOKIE, buf.toString());
563 }
564 }
565
566
567 if (_authorizations != null)
568 {
569 Authentication auth = (Authentication)_authorizations.match(ex.getRequestURI());
570 if (auth != null)
571 (auth).setCredentials(ex);
572 }
573
574
575
576 ex.scheduleTimeout(this);
577
578 AbstractHttpConnection connection = getIdleConnection();
579 if (connection != null)
580 {
581 send(connection, ex);
582 }
583 else
584 {
585 boolean startConnection = false;
586 synchronized (this)
587 {
588 if (_exchanges.size() == _maxQueueSize)
589 throw new RejectedExecutionException("Queue full for address " + _address);
590
591 _exchanges.add(ex);
592 if (_connections.size() + _pendingConnections < _maxConnections)
593 startConnection = true;
594 }
595
596 if (startConnection)
597 startNewConnection();
598 }
599 }
600
601 protected void exchangeExpired(HttpExchange exchange)
602 {
603
604
605 synchronized (this)
606 {
607 _exchanges.remove(exchange);
608 }
609 }
610
611 protected void send(AbstractHttpConnection connection, HttpExchange exchange) throws IOException
612 {
613 synchronized (this)
614 {
615
616
617 if (!connection.send(exchange))
618 {
619 if (exchange.getStatus() <= HttpExchange.STATUS_WAITING_FOR_CONNECTION)
620 _exchanges.add(0, exchange);
621 returnIdleConnection(connection);
622 }
623 }
624 }
625
626 @Override
627 public synchronized String toString()
628 {
629 return String.format("HttpDestination@%x//%s:%d(%d/%d,%d,%d/%d)%n", hashCode(), _address.getHost(), _address.getPort(), _connections.size(), _maxConnections, _idleConnections.size(), _exchanges.size(), _maxQueueSize);
630 }
631
632 public synchronized String toDetailString()
633 {
634 StringBuilder b = new StringBuilder();
635 b.append(toString());
636 b.append('\n');
637 synchronized (this)
638 {
639 for (AbstractHttpConnection connection : _connections)
640 {
641 b.append(connection.toDetailString());
642 if (_idleConnections.contains(connection))
643 b.append(" IDLE");
644 b.append('\n');
645 }
646 }
647 b.append("--");
648 b.append('\n');
649
650 return b.toString();
651 }
652
653 public void setProxy(Address proxy)
654 {
655 _proxy = proxy;
656 }
657
658 public Address getProxy()
659 {
660 return _proxy;
661 }
662
663 public Authentication getProxyAuthentication()
664 {
665 return _proxyAuthentication;
666 }
667
668 public void setProxyAuthentication(Authentication authentication)
669 {
670 _proxyAuthentication = authentication;
671 }
672
673 public boolean isProxied()
674 {
675 return _proxy != null;
676 }
677
678 public void close() throws IOException
679 {
680 synchronized (this)
681 {
682 for (AbstractHttpConnection connection : _connections)
683 {
684 connection.close();
685 }
686 }
687 }
688
689 public String dump()
690 {
691 return AggregateLifeCycle.dump(this);
692 }
693
694 public void dump(Appendable out, String indent) throws IOException
695 {
696 synchronized (this)
697 {
698 out.append(String.valueOf(this));
699 out.append("idle=");
700 out.append(String.valueOf(_idleConnections.size()));
701 out.append(" pending=");
702 out.append(String.valueOf(_pendingConnections));
703 out.append("\n");
704 AggregateLifeCycle.dump(out, indent, _connections);
705 }
706 }
707
708 private class ConnectExchange extends ContentExchange
709 {
710 private final SelectConnector.UpgradableEndPoint proxyEndPoint;
711
712 public ConnectExchange(Address serverAddress, SelectConnector.UpgradableEndPoint proxyEndPoint)
713 {
714 this.proxyEndPoint = proxyEndPoint;
715 setMethod(HttpMethods.CONNECT);
716 String serverHostAndPort = serverAddress.toString();
717 setRequestURI(serverHostAndPort);
718 addRequestHeader(HttpHeaders.HOST, serverHostAndPort);
719 addRequestHeader(HttpHeaders.PROXY_CONNECTION, "keep-alive");
720 addRequestHeader(HttpHeaders.USER_AGENT, "Jetty-Client");
721 }
722
723 @Override
724 protected void onResponseComplete() throws IOException
725 {
726 int responseStatus = getResponseStatus();
727 if (responseStatus == HttpStatus.OK_200)
728 {
729 proxyEndPoint.upgrade();
730 }
731 else if (responseStatus == HttpStatus.GATEWAY_TIMEOUT_504)
732 {
733 onExpire();
734 }
735 else
736 {
737 onException(new ProtocolException("Proxy: " + proxyEndPoint.getRemoteAddr() + ":" + proxyEndPoint.getRemotePort() + " didn't return http return code 200, but " + responseStatus));
738 }
739 }
740
741 @Override
742 protected void onConnectionFailed(Throwable x)
743 {
744 HttpDestination.this.onConnectionFailed(x);
745 }
746
747 @Override
748 protected void onException(Throwable x)
749 {
750 HttpExchange exchange = null;
751 synchronized (HttpDestination.this)
752 {
753 if (!_exchanges.isEmpty())
754 exchange = _exchanges.remove(0);
755 }
756 if (exchange != null && exchange.setStatus(STATUS_EXCEPTED))
757 exchange.getEventListener().onException(x);
758 }
759
760 @Override
761 protected void onExpire()
762 {
763 HttpExchange exchange = null;
764 synchronized (HttpDestination.this)
765 {
766 if (!_exchanges.isEmpty())
767 exchange = _exchanges.remove(0);
768 }
769 if (exchange != null && exchange.setStatus(STATUS_EXPIRED))
770 exchange.getEventListener().onExpire();
771 }
772 }
773 }