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.io.InputStream;
23 import java.util.Collections;
24 import java.util.concurrent.atomic.AtomicBoolean;
25
26 import org.eclipse.jetty.client.security.Authentication;
27 import org.eclipse.jetty.http.HttpFields;
28 import org.eclipse.jetty.http.HttpGenerator;
29 import org.eclipse.jetty.http.HttpHeaderValues;
30 import org.eclipse.jetty.http.HttpHeaders;
31 import org.eclipse.jetty.http.HttpMethods;
32 import org.eclipse.jetty.http.HttpParser;
33 import org.eclipse.jetty.http.HttpSchemes;
34 import org.eclipse.jetty.http.HttpStatus;
35 import org.eclipse.jetty.http.HttpVersions;
36 import org.eclipse.jetty.io.AbstractConnection;
37 import org.eclipse.jetty.io.Buffer;
38 import org.eclipse.jetty.io.Buffers;
39 import org.eclipse.jetty.io.Connection;
40 import org.eclipse.jetty.io.EndPoint;
41 import org.eclipse.jetty.io.EofException;
42 import org.eclipse.jetty.io.View;
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.thread.Timeout;
48
49
50
51
52
53 public abstract class AbstractHttpConnection extends AbstractConnection implements Dumpable
54 {
55 private static final Logger LOG = Log.getLogger(AbstractHttpConnection.class);
56
57 protected HttpDestination _destination;
58 protected HttpGenerator _generator;
59 protected HttpParser _parser;
60 protected boolean _http11 = true;
61 protected int _status;
62 protected Buffer _connectionHeader;
63 protected boolean _reserved;
64
65
66 protected volatile HttpExchange _exchange;
67 protected HttpExchange _pipeline;
68 private final Timeout.Task _idleTimeout = new ConnectionIdleTask();
69 private AtomicBoolean _idle = new AtomicBoolean(false);
70
71
72 AbstractHttpConnection(Buffers requestBuffers, Buffers responseBuffers, EndPoint endp)
73 {
74 super(endp);
75
76 _generator = new HttpGenerator(requestBuffers,endp);
77 _parser = new HttpParser(responseBuffers,endp,new Handler());
78 }
79
80 public void setReserved (boolean reserved)
81 {
82 _reserved = reserved;
83 }
84
85 public boolean isReserved()
86 {
87 return _reserved;
88 }
89
90 public HttpDestination getDestination()
91 {
92 return _destination;
93 }
94
95 public void setDestination(HttpDestination destination)
96 {
97 _destination = destination;
98 }
99
100 public boolean send(HttpExchange ex) throws IOException
101 {
102 LOG.debug("Send {} on {}",ex,this);
103 synchronized (this)
104 {
105 if (_exchange != null)
106 {
107 if (_pipeline != null)
108 throw new IllegalStateException(this + " PIPELINED!!! _exchange=" + _exchange);
109 _pipeline = ex;
110 return true;
111 }
112
113 _exchange = ex;
114 _exchange.associate(this);
115
116
117 if (!_endp.isOpen())
118 {
119 _exchange.disassociate();
120 _exchange = null;
121 return false;
122 }
123
124 _exchange.setStatus(HttpExchange.STATUS_WAITING_FOR_COMMIT);
125
126 adjustIdleTimeout();
127
128 return true;
129 }
130 }
131
132 private void adjustIdleTimeout() throws IOException
133 {
134
135
136
137
138
139 long timeout = _exchange.getTimeout();
140 if (timeout <= 0)
141 timeout = _destination.getHttpClient().getTimeout();
142
143 long endPointTimeout = _endp.getMaxIdleTime();
144
145 if (timeout > 0 && timeout > endPointTimeout)
146 {
147
148
149
150 _endp.setMaxIdleTime(2 * (int)timeout);
151 }
152 }
153
154 public abstract Connection handle() throws IOException;
155
156
157 public boolean isIdle()
158 {
159 synchronized (this)
160 {
161 return _exchange == null;
162 }
163 }
164
165 public boolean isSuspended()
166 {
167 return false;
168 }
169
170 public void onClose()
171 {
172 }
173
174
175
176
177 protected void commitRequest() throws IOException
178 {
179 synchronized (this)
180 {
181 _status=0;
182 if (_exchange.getStatus() != HttpExchange.STATUS_WAITING_FOR_COMMIT)
183 throw new IllegalStateException();
184
185 _exchange.setStatus(HttpExchange.STATUS_SENDING_REQUEST);
186 _generator.setVersion(_exchange.getVersion());
187
188 String method=_exchange.getMethod();
189 String uri = _exchange.getRequestURI();
190 if (_destination.isProxied())
191 {
192 if (!HttpMethods.CONNECT.equals(method) && uri.startsWith("/"))
193 {
194 boolean secure = _destination.isSecure();
195 String host = _destination.getAddress().getHost();
196 int port = _destination.getAddress().getPort();
197 StringBuilder absoluteURI = new StringBuilder();
198 absoluteURI.append(secure ? HttpSchemes.HTTPS : HttpSchemes.HTTP);
199 absoluteURI.append("://");
200 absoluteURI.append(host);
201
202 if (!(secure && port == 443 || !secure && port == 80))
203 absoluteURI.append(":").append(port);
204 absoluteURI.append(uri);
205 uri = absoluteURI.toString();
206 }
207 Authentication auth = _destination.getProxyAuthentication();
208 if (auth != null)
209 auth.setCredentials(_exchange);
210 }
211
212 _generator.setRequest(method, uri);
213 _parser.setHeadResponse(HttpMethods.HEAD.equalsIgnoreCase(method));
214
215 HttpFields requestHeaders = _exchange.getRequestFields();
216 if (_exchange.getVersion() >= HttpVersions.HTTP_1_1_ORDINAL)
217 {
218 if (!requestHeaders.containsKey(HttpHeaders.HOST_BUFFER))
219 requestHeaders.add(HttpHeaders.HOST_BUFFER,_destination.getHostHeader());
220 }
221
222 Buffer requestContent = _exchange.getRequestContent();
223 if (requestContent != null)
224 {
225 requestHeaders.putLongField(HttpHeaders.CONTENT_LENGTH, requestContent.length());
226 _generator.completeHeader(requestHeaders,false);
227 _generator.addContent(new View(requestContent),true);
228 _exchange.setStatus(HttpExchange.STATUS_WAITING_FOR_RESPONSE);
229 }
230 else
231 {
232 InputStream requestContentStream = _exchange.getRequestContentSource();
233 if (requestContentStream != null)
234 {
235 _generator.completeHeader(requestHeaders, false);
236 }
237 else
238 {
239 requestHeaders.remove(HttpHeaders.CONTENT_LENGTH);
240 _generator.completeHeader(requestHeaders, true);
241 _exchange.setStatus(HttpExchange.STATUS_WAITING_FOR_RESPONSE);
242 }
243 }
244 }
245 }
246
247 protected void reset() throws IOException
248 {
249 _connectionHeader = null;
250 _parser.reset();
251 _generator.reset();
252 _http11 = true;
253 }
254
255
256 private class Handler extends HttpParser.EventHandler
257 {
258 @Override
259 public void startRequest(Buffer method, Buffer url, Buffer version) throws IOException
260 {
261
262
263
264
265
266 }
267
268 @Override
269 public void startResponse(Buffer version, int status, Buffer reason) throws IOException
270 {
271 HttpExchange exchange = _exchange;
272 if (exchange==null)
273 {
274 LOG.warn("No exchange for response");
275 _endp.close();
276 return;
277 }
278
279 switch(status)
280 {
281 case HttpStatus.CONTINUE_100:
282 case HttpStatus.PROCESSING_102:
283
284 exchange.setEventListener(new NonFinalResponseListener(exchange));
285 break;
286
287 case HttpStatus.OK_200:
288
289 if (HttpMethods.CONNECT.equalsIgnoreCase(exchange.getMethod()))
290 _parser.setHeadResponse(true);
291 break;
292 }
293
294 _http11 = HttpVersions.HTTP_1_1_BUFFER.equals(version);
295 _status=status;
296 exchange.getEventListener().onResponseStatus(version,status,reason);
297 exchange.setStatus(HttpExchange.STATUS_PARSING_HEADERS);
298
299 }
300
301 @Override
302 public void parsedHeader(Buffer name, Buffer value) throws IOException
303 {
304 HttpExchange exchange = _exchange;
305 if (exchange!=null)
306 {
307 if (HttpHeaders.CACHE.getOrdinal(name) == HttpHeaders.CONNECTION_ORDINAL)
308 {
309 _connectionHeader = HttpHeaderValues.CACHE.lookup(value);
310 }
311 exchange.getEventListener().onResponseHeader(name,value);
312 }
313 }
314
315 @Override
316 public void headerComplete() throws IOException
317 {
318 HttpExchange exchange = _exchange;
319 if (exchange!=null)
320 {
321 exchange.setStatus(HttpExchange.STATUS_PARSING_CONTENT);
322 if (HttpMethods.CONNECT.equalsIgnoreCase(exchange.getMethod()))
323 _parser.setPersistent(true);
324 }
325 }
326
327 @Override
328 public void content(Buffer ref) throws IOException
329 {
330 HttpExchange exchange = _exchange;
331 if (exchange!=null)
332 exchange.getEventListener().onResponseContent(ref);
333 }
334
335 @Override
336 public void messageComplete(long contextLength) throws IOException
337 {
338 HttpExchange exchange = _exchange;
339 if (exchange!=null)
340 exchange.setStatus(HttpExchange.STATUS_COMPLETED);
341 }
342
343 @Override
344 public void earlyEOF()
345 {
346 HttpExchange exchange = _exchange;
347 if (exchange!=null)
348 {
349 if (!exchange.isDone())
350 {
351 if (exchange.setStatus(HttpExchange.STATUS_EXCEPTED))
352 exchange.getEventListener().onException(new EofException("early EOF"));
353 }
354 }
355 }
356 }
357
358 @Override
359 public String toString()
360 {
361 return String.format("%s %s g=%s p=%s",
362 super.toString(),
363 _destination == null ? "?.?.?.?:??" : _destination.getAddress(),
364 _generator,
365 _parser);
366 }
367
368 public String toDetailString()
369 {
370 return toString() + " ex=" + _exchange + " idle for " + _idleTimeout.getAge();
371 }
372
373 public void close() throws IOException
374 {
375
376
377
378 HttpExchange exchange = _exchange;
379 if (exchange != null && !exchange.isDone())
380 {
381 switch (exchange.getStatus())
382 {
383 case HttpExchange.STATUS_CANCELLED:
384 case HttpExchange.STATUS_CANCELLING:
385 case HttpExchange.STATUS_COMPLETED:
386 case HttpExchange.STATUS_EXCEPTED:
387 case HttpExchange.STATUS_EXPIRED:
388 break;
389 case HttpExchange.STATUS_PARSING_CONTENT:
390 if (_endp.isInputShutdown() && _parser.isState(HttpParser.STATE_EOF_CONTENT))
391 break;
392 default:
393 String exch= exchange.toString();
394 String reason = _endp.isOpen()?(_endp.isInputShutdown()?"half closed: ":"local close: "):"closed: ";
395 if (exchange.setStatus(HttpExchange.STATUS_EXCEPTED))
396 exchange.getEventListener().onException(new EofException(reason+exch));
397 }
398 }
399
400 if (_endp.isOpen())
401 {
402 _endp.close();
403 _destination.returnConnection(this, true);
404 }
405 }
406
407 public void setIdleTimeout()
408 {
409 synchronized (this)
410 {
411 if (_idle.compareAndSet(false, true))
412 _destination.getHttpClient().scheduleIdle(_idleTimeout);
413 else
414 throw new IllegalStateException();
415 }
416 }
417
418 public boolean cancelIdleTimeout()
419 {
420 synchronized (this)
421 {
422 if (_idle.compareAndSet(true, false))
423 {
424 _destination.getHttpClient().cancel(_idleTimeout);
425 return true;
426 }
427 }
428
429 return false;
430 }
431
432 protected void exchangeExpired(HttpExchange exchange)
433 {
434 synchronized (this)
435 {
436
437
438 if (_exchange == exchange)
439 {
440 try
441 {
442 _destination.returnConnection(this, true);
443 }
444 catch (IOException x)
445 {
446 LOG.ignore(x);
447 }
448 }
449 }
450 }
451
452
453
454
455
456 public String dump()
457 {
458 return AggregateLifeCycle.dump(this);
459 }
460
461
462
463
464
465 public void dump(Appendable out, String indent) throws IOException
466 {
467 synchronized (this)
468 {
469 out.append(String.valueOf(this)).append("\n");
470 AggregateLifeCycle.dump(out,indent,Collections.singletonList(_endp));
471 }
472 }
473
474
475 private class ConnectionIdleTask extends Timeout.Task
476 {
477
478 @Override
479 public void expired()
480 {
481
482 if (_idle.compareAndSet(true, false))
483 {
484 _destination.returnIdleConnection(AbstractHttpConnection.this);
485 }
486 }
487 }
488
489
490
491 private class NonFinalResponseListener implements HttpEventListener
492 {
493 final HttpExchange _exchange;
494 final HttpEventListener _next;
495
496
497 public NonFinalResponseListener(HttpExchange exchange)
498 {
499 _exchange=exchange;
500 _next=exchange.getEventListener();
501 }
502
503
504 public void onRequestCommitted() throws IOException
505 {
506 }
507
508
509 public void onRequestComplete() throws IOException
510 {
511 }
512
513
514 public void onResponseStatus(Buffer version, int status, Buffer reason) throws IOException
515 {
516 }
517
518
519 public void onResponseHeader(Buffer name, Buffer value) throws IOException
520 {
521 _next.onResponseHeader(name,value);
522 }
523
524
525 public void onResponseHeaderComplete() throws IOException
526 {
527 _next.onResponseHeaderComplete();
528 }
529
530
531 public void onResponseContent(Buffer content) throws IOException
532 {
533 }
534
535
536 public void onResponseComplete() throws IOException
537 {
538 _exchange.setEventListener(_next);
539 _exchange.setStatus(HttpExchange.STATUS_WAITING_FOR_RESPONSE);
540 _parser.reset();
541 }
542
543
544 public void onConnectionFailed(Throwable ex)
545 {
546 _exchange.setEventListener(_next);
547 _next.onConnectionFailed(ex);
548 }
549
550
551 public void onException(Throwable ex)
552 {
553 _exchange.setEventListener(_next);
554 _next.onException(ex);
555 }
556
557
558 public void onExpire()
559 {
560 _exchange.setEventListener(_next);
561 _next.onExpire();
562 }
563
564
565 public void onRetry()
566 {
567 _exchange.setEventListener(_next);
568 _next.onRetry();
569 }
570 }
571 }