1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.server;
20
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.nio.channels.WritePendingException;
24 import java.util.concurrent.RejectedExecutionException;
25 import java.util.concurrent.atomic.AtomicInteger;
26
27 import org.eclipse.jetty.http.HttpCompliance;
28 import org.eclipse.jetty.http.HttpField;
29 import org.eclipse.jetty.http.HttpGenerator;
30 import org.eclipse.jetty.http.HttpHeader;
31 import org.eclipse.jetty.http.HttpHeaderValue;
32 import org.eclipse.jetty.http.HttpParser;
33 import org.eclipse.jetty.http.HttpParser.RequestHandler;
34 import org.eclipse.jetty.http.HttpStatus;
35 import org.eclipse.jetty.http.MetaData;
36 import org.eclipse.jetty.http.PreEncodedHttpField;
37 import org.eclipse.jetty.io.AbstractConnection;
38 import org.eclipse.jetty.io.ByteBufferPool;
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.util.BufferUtil;
43 import org.eclipse.jetty.util.Callback;
44 import org.eclipse.jetty.util.IteratingCallback;
45 import org.eclipse.jetty.util.log.Log;
46 import org.eclipse.jetty.util.log.Logger;
47
48
49
50
51 public class HttpConnection extends AbstractConnection implements Runnable, HttpTransport, Connection.UpgradeFrom
52 {
53 private static final Logger LOG = Log.getLogger(HttpConnection.class);
54 public static final HttpField CONNECTION_CLOSE = new PreEncodedHttpField(HttpHeader.CONNECTION,HttpHeaderValue.CLOSE.asString());
55 public static final String UPGRADE_CONNECTION_ATTRIBUTE = "org.eclipse.jetty.server.HttpConnection.UPGRADE";
56 private static final boolean REQUEST_BUFFER_DIRECT=false;
57 private static final boolean HEADER_BUFFER_DIRECT=false;
58 private static final boolean CHUNK_BUFFER_DIRECT=false;
59 private static final ThreadLocal<HttpConnection> __currentConnection = new ThreadLocal<>();
60
61 private final HttpConfiguration _config;
62 private final Connector _connector;
63 private final ByteBufferPool _bufferPool;
64 private final HttpInput _input;
65 private final HttpGenerator _generator;
66 private final HttpChannelOverHttp _channel;
67 private final HttpParser _parser;
68 private final AtomicInteger _contentBufferReferences=new AtomicInteger();
69 private volatile ByteBuffer _requestBuffer = null;
70 private volatile ByteBuffer _chunk = null;
71 private final BlockingReadCallback _blockingReadCallback = new BlockingReadCallback();
72 private final AsyncReadCallback _asyncReadCallback = new AsyncReadCallback();
73 private final SendCallback _sendCallback = new SendCallback();
74 private final boolean _recordHttpComplianceViolations;
75
76
77
78
79
80
81
82
83 public static HttpConnection getCurrentConnection()
84 {
85 return __currentConnection.get();
86 }
87
88 protected static HttpConnection setCurrentConnection(HttpConnection connection)
89 {
90 HttpConnection last=__currentConnection.get();
91 __currentConnection.set(connection);
92 return last;
93 }
94
95 public HttpConnection(HttpConfiguration config, Connector connector, EndPoint endPoint, HttpCompliance compliance, boolean recordComplianceViolations)
96 {
97 super(endPoint, connector.getExecutor());
98 _config = config;
99 _connector = connector;
100 _bufferPool = _connector.getByteBufferPool();
101 _generator = newHttpGenerator();
102 _channel = newHttpChannel();
103 _input = _channel.getRequest().getHttpInput();
104 _parser = newHttpParser(compliance);
105 _recordHttpComplianceViolations=recordComplianceViolations;
106 if (LOG.isDebugEnabled())
107 LOG.debug("New HTTP Connection {}", this);
108 }
109
110 public HttpConfiguration getHttpConfiguration()
111 {
112 return _config;
113 }
114
115 public boolean isRecordHttpComplianceViolations()
116 {
117 return _recordHttpComplianceViolations;
118 }
119
120 protected HttpGenerator newHttpGenerator()
121 {
122 return new HttpGenerator(_config.getSendServerVersion(),_config.getSendXPoweredBy());
123 }
124
125 protected HttpChannelOverHttp newHttpChannel()
126 {
127 HttpChannelOverHttp httpChannel = new HttpChannelOverHttp(this, _connector, _config, getEndPoint(), this);
128
129 return httpChannel;
130 }
131
132 protected HttpParser newHttpParser(HttpCompliance compliance)
133 {
134 return new HttpParser(newRequestHandler(), getHttpConfiguration().getRequestHeaderSize(), compliance);
135 }
136
137 protected HttpParser.RequestHandler newRequestHandler()
138 {
139 return _channel;
140 }
141
142 public Server getServer()
143 {
144 return _connector.getServer();
145 }
146
147 public Connector getConnector()
148 {
149 return _connector;
150 }
151
152 public HttpChannel getHttpChannel()
153 {
154 return _channel;
155 }
156
157 public HttpParser getParser()
158 {
159 return _parser;
160 }
161
162 public HttpGenerator getGenerator()
163 {
164 return _generator;
165 }
166
167 @Override
168 public boolean isOptimizedForDirectBuffers()
169 {
170 return getEndPoint().isOptimizedForDirectBuffers();
171 }
172
173 @Override
174 public int getMessagesIn()
175 {
176 return getHttpChannel().getRequests();
177 }
178
179 @Override
180 public int getMessagesOut()
181 {
182 return getHttpChannel().getRequests();
183 }
184
185 @Override
186 public ByteBuffer onUpgradeFrom()
187 {
188 if (BufferUtil.hasContent(_requestBuffer))
189 {
190 ByteBuffer buffer = _requestBuffer;
191 _requestBuffer=null;
192 return buffer;
193 }
194 return null;
195 }
196
197 void releaseRequestBuffer()
198 {
199 if (_requestBuffer != null && !_requestBuffer.hasRemaining())
200 {
201 if (LOG.isDebugEnabled())
202 LOG.debug("releaseRequestBuffer {}",this);
203 ByteBuffer buffer=_requestBuffer;
204 _requestBuffer=null;
205 _bufferPool.release(buffer);
206 }
207 }
208
209 public ByteBuffer getRequestBuffer()
210 {
211 if (_requestBuffer == null)
212 _requestBuffer = _bufferPool.acquire(getInputBufferSize(), REQUEST_BUFFER_DIRECT);
213 return _requestBuffer;
214 }
215
216 public boolean isRequestBufferEmpty()
217 {
218 return BufferUtil.isEmpty(_requestBuffer);
219 }
220
221 @Override
222 public void onFillable()
223 {
224 if (LOG.isDebugEnabled())
225 LOG.debug("{} onFillable enter {} {}", this, _channel.getState(),BufferUtil.toDetailString(_requestBuffer));
226
227 HttpConnection last=setCurrentConnection(this);
228 try
229 {
230 while (getEndPoint().isOpen())
231 {
232
233 int filled = fillRequestBuffer();
234
235
236 boolean handle = parseRequestBuffer();
237
238
239
240 if (getEndPoint().getConnection()!=this)
241 break;
242
243
244 if (_parser.isClose() || _parser.isClosed())
245 {
246 close();
247 break;
248 }
249
250
251 if (handle)
252 {
253 boolean suspended = !_channel.handle();
254
255
256 if (suspended || getEndPoint().getConnection() != this)
257 break;
258 }
259 else
260 {
261 if (filled <= 0)
262 {
263 if (filled == 0)
264 fillInterested();
265 break;
266 }
267 }
268 }
269 }
270 finally
271 {
272 setCurrentConnection(last);
273 if (LOG.isDebugEnabled())
274 LOG.debug("{} onFillable exit {} {}", this, _channel.getState(),BufferUtil.toDetailString(_requestBuffer));
275 }
276 }
277
278
279
280
281
282 protected boolean fillAndParseForContent()
283 {
284 boolean handled=false;
285 while (_parser.inContentState())
286 {
287 int filled = fillRequestBuffer();
288 boolean handle = parseRequestBuffer();
289 handled|=handle;
290 if (handle || filled<=0 || _channel.getRequest().getHttpInput().hasContent())
291 break;
292 }
293 return handled;
294 }
295
296
297 private int fillRequestBuffer()
298 {
299 if (_contentBufferReferences.get()>0)
300 {
301 LOG.warn("{} fill with unconsumed content!",this);
302 return 0;
303 }
304
305 if (BufferUtil.isEmpty(_requestBuffer))
306 {
307
308 if(getEndPoint().isInputShutdown())
309 {
310
311 _parser.atEOF();
312 if (LOG.isDebugEnabled())
313 LOG.debug("{} filled -1 {}",this,BufferUtil.toDetailString(_requestBuffer));
314 return -1;
315 }
316
317
318
319
320 _requestBuffer = getRequestBuffer();
321
322
323 try
324 {
325 int filled = getEndPoint().fill(_requestBuffer);
326 if (filled==0)
327 filled = getEndPoint().fill(_requestBuffer);
328
329
330 if (filled < 0)
331 _parser.atEOF();
332
333 if (LOG.isDebugEnabled())
334 LOG.debug("{} filled {} {}",this,filled,BufferUtil.toDetailString(_requestBuffer));
335
336 return filled;
337 }
338 catch (IOException e)
339 {
340 LOG.debug(e);
341 return -1;
342 }
343 }
344 return 0;
345 }
346
347
348 private boolean parseRequestBuffer()
349 {
350 if (LOG.isDebugEnabled())
351 LOG.debug("{} parse {} {}",this,BufferUtil.toDetailString(_requestBuffer));
352
353 boolean handle = _parser.parseNext(_requestBuffer==null?BufferUtil.EMPTY_BUFFER:_requestBuffer);
354
355 if (LOG.isDebugEnabled())
356 LOG.debug("{} parsed {} {}",this,handle,_parser);
357
358
359 if (_contentBufferReferences.get()==0)
360 releaseRequestBuffer();
361
362 return handle;
363 }
364
365
366 @Override
367 public void onCompleted()
368 {
369
370 if (_channel.getResponse().getStatus() == HttpStatus.SWITCHING_PROTOCOLS_101)
371 {
372 Connection connection = (Connection)_channel.getRequest().getAttribute(UPGRADE_CONNECTION_ATTRIBUTE);
373 if (connection != null)
374 {
375 if (LOG.isDebugEnabled())
376 LOG.debug("Upgrade from {} to {}", this, connection);
377 _channel.getState().upgrade();
378 getEndPoint().upgrade(connection);
379 _channel.recycle();
380 _parser.reset();
381 _generator.reset();
382 if (_contentBufferReferences.get()==0)
383 releaseRequestBuffer();
384 else
385 {
386 LOG.warn("{} lingering content references?!?!",this);
387 _requestBuffer=null;
388 _contentBufferReferences.set(0);
389 }
390 return;
391 }
392 }
393
394
395
396 if (_channel.isExpecting100Continue())
397 {
398
399 _parser.close();
400 }
401 else if (_parser.inContentState() && _generator.isPersistent())
402 {
403
404 if (_channel.getRequest().getHttpInput().isAsync())
405 {
406 if (LOG.isDebugEnabled())
407 LOG.debug("unconsumed async input {}", this);
408 _channel.abort(new IOException("unconsumed input"));
409 }
410 else
411 {
412 if (LOG.isDebugEnabled())
413 LOG.debug("unconsumed input {}", this);
414
415 if (!_channel.getRequest().getHttpInput().consumeAll())
416 _channel.abort(new IOException("unconsumed input"));
417 }
418 }
419
420
421 _channel.recycle();
422 if (_generator.isPersistent() && !_parser.isClosed())
423 _parser.reset();
424 else
425 _parser.close();
426
427
428
429 if (_chunk!=null)
430 _bufferPool.release(_chunk);
431 _chunk=null;
432 _generator.reset();
433
434
435 if (getCurrentConnection()!=this)
436 {
437
438 if (_parser.isStart())
439 {
440
441 if (BufferUtil.isEmpty(_requestBuffer))
442 {
443
444 fillInterested();
445 }
446
447 else if (getConnector().isRunning())
448 {
449
450 try
451 {
452 getExecutor().execute(this);
453 }
454 catch (RejectedExecutionException e)
455 {
456 if (getConnector().isRunning())
457 LOG.warn(e);
458 else
459 LOG.ignore(e);
460 getEndPoint().close();
461 }
462 }
463 else
464 {
465 getEndPoint().close();
466 }
467 }
468
469 else if (getEndPoint().isOpen())
470 fillInterested();
471 }
472 }
473
474 @Override
475 protected void onFillInterestedFailed(Throwable cause)
476 {
477 _parser.close();
478 super.onFillInterestedFailed(cause);
479 }
480
481 @Override
482 public void onOpen()
483 {
484 super.onOpen();
485 fillInterested();
486 }
487
488 @Override
489 public void onClose()
490 {
491 _sendCallback.close();
492 super.onClose();
493 }
494
495 @Override
496 public void run()
497 {
498 onFillable();
499 }
500
501 @Override
502 public void send(MetaData.Response info, boolean head, ByteBuffer content, boolean lastContent, Callback callback)
503 {
504 if (info == null)
505 {
506 if (!lastContent && BufferUtil.isEmpty(content))
507 {
508 callback.succeeded();
509 return;
510 }
511 }
512 else
513 {
514
515 if (_channel.isExpecting100Continue())
516
517 _generator.setPersistent(false);
518 }
519
520 if(_sendCallback.reset(info,head,content,lastContent,callback))
521 _sendCallback.iterate();
522 }
523
524
525 HttpInput.Content newContent(ByteBuffer c)
526 {
527 return new Content(c);
528 }
529
530 @Override
531 public void abort(Throwable failure)
532 {
533
534
535 getEndPoint().close();
536 }
537
538 @Override
539 public boolean isPushSupported()
540 {
541 return false;
542 }
543
544 @Override
545 public void push(org.eclipse.jetty.http.MetaData.Request request)
546 {
547 LOG.debug("ignore push in {}",this);
548 }
549
550 public void asyncReadFillInterested()
551 {
552 getEndPoint().fillInterested(_asyncReadCallback);
553 }
554
555 public void blockingReadFillInterested()
556 {
557 getEndPoint().fillInterested(_blockingReadCallback);
558 }
559
560 public void blockingReadException(Throwable e)
561 {
562 _blockingReadCallback.failed(e);
563 }
564
565 @Override
566 public String toString()
567 {
568 return String.format("%s[p=%s,g=%s,c=%s]",
569 super.toString(),
570 _parser,
571 _generator,
572 _channel);
573 }
574
575 private class Content extends HttpInput.Content
576 {
577 public Content(ByteBuffer content)
578 {
579 super(content);
580 _contentBufferReferences.incrementAndGet();
581 }
582
583 @Override
584 public void succeeded()
585 {
586 if (_contentBufferReferences.decrementAndGet()==0)
587 releaseRequestBuffer();
588 }
589
590 @Override
591 public void failed(Throwable x)
592 {
593 succeeded();
594 }
595 }
596
597 private class BlockingReadCallback implements Callback
598 {
599 @Override
600 public void succeeded()
601 {
602 _input.unblock();
603 }
604
605 @Override
606 public void failed(Throwable x)
607 {
608 _input.failed(x);
609 }
610
611 @Override
612 public boolean isNonBlocking()
613 {
614
615
616 return true;
617 }
618 }
619
620 private class AsyncReadCallback implements Callback
621 {
622 @Override
623 public void succeeded()
624 {
625 if (fillAndParseForContent())
626 _channel.handle();
627 else if (!_input.isFinished())
628 asyncReadFillInterested();
629 }
630
631 @Override
632 public void failed(Throwable x)
633 {
634 if (_input.failed(x))
635 _channel.handle();
636 }
637 }
638
639 private class SendCallback extends IteratingCallback
640 {
641 private MetaData.Response _info;
642 private boolean _head;
643 private ByteBuffer _content;
644 private boolean _lastContent;
645 private Callback _callback;
646 private ByteBuffer _header;
647 private boolean _shutdownOut;
648
649 private SendCallback()
650 {
651 super(true);
652 }
653
654 @Override
655 public boolean isNonBlocking()
656 {
657 return _callback.isNonBlocking();
658 }
659
660 private boolean reset(MetaData.Response info, boolean head, ByteBuffer content, boolean last, Callback callback)
661 {
662 if (reset())
663 {
664 _info = info;
665 _head = head;
666 _content = content;
667 _lastContent = last;
668 _callback = callback;
669 _header = null;
670 _shutdownOut = false;
671 return true;
672 }
673
674 if (isClosed())
675 callback.failed(new EofException());
676 else
677 callback.failed(new WritePendingException());
678 return false;
679 }
680
681 @Override
682 public Action process() throws Exception
683 {
684 if (_callback==null)
685 throw new IllegalStateException();
686
687 ByteBuffer chunk = _chunk;
688 while (true)
689 {
690 HttpGenerator.Result result = _generator.generateResponse(_info, _head, _header, chunk, _content, _lastContent);
691 if (LOG.isDebugEnabled())
692 LOG.debug("{} generate: {} ({},{},{})@{}",
693 this,
694 result,
695 BufferUtil.toSummaryString(_header),
696 BufferUtil.toSummaryString(_content),
697 _lastContent,
698 _generator.getState());
699
700 switch (result)
701 {
702 case NEED_INFO:
703 throw new EofException("request lifecycle violation");
704
705 case NEED_HEADER:
706 {
707 _header = _bufferPool.acquire(_config.getResponseHeaderSize(), HEADER_BUFFER_DIRECT);
708
709 continue;
710 }
711 case NEED_CHUNK:
712 {
713 chunk = _chunk = _bufferPool.acquire(HttpGenerator.CHUNK_SIZE, CHUNK_BUFFER_DIRECT);
714 continue;
715 }
716 case FLUSH:
717 {
718
719 if (_head || _generator.isNoContent())
720 {
721 BufferUtil.clear(chunk);
722 BufferUtil.clear(_content);
723 }
724
725
726 if (BufferUtil.hasContent(_header))
727 {
728 if (BufferUtil.hasContent(_content))
729 {
730 if (BufferUtil.hasContent(chunk))
731 getEndPoint().write(this, _header, chunk, _content);
732 else
733 getEndPoint().write(this, _header, _content);
734 }
735 else
736 getEndPoint().write(this, _header);
737 }
738 else if (BufferUtil.hasContent(chunk))
739 {
740 if (BufferUtil.hasContent(_content))
741 getEndPoint().write(this, chunk, _content);
742 else
743 getEndPoint().write(this, chunk);
744 }
745 else if (BufferUtil.hasContent(_content))
746 {
747 getEndPoint().write(this, _content);
748 }
749 else
750 {
751 succeeded();
752 }
753 return Action.SCHEDULED;
754 }
755 case SHUTDOWN_OUT:
756 {
757 _shutdownOut=true;
758 continue;
759 }
760 case DONE:
761 {
762 return Action.SUCCEEDED;
763 }
764 case CONTINUE:
765 {
766 break;
767 }
768 default:
769 {
770 throw new IllegalStateException("generateResponse="+result);
771 }
772 }
773 }
774 }
775
776 private void releaseHeader()
777 {
778 ByteBuffer h=_header;
779 _header=null;
780 if (h!=null)
781 _bufferPool.release(h);
782 }
783
784 @Override
785 protected void onCompleteSuccess()
786 {
787 releaseHeader();
788 _callback.succeeded();
789 if (_shutdownOut)
790 getEndPoint().shutdownOutput();
791 }
792
793 @Override
794 public void onCompleteFailure(final Throwable x)
795 {
796 releaseHeader();
797 failedCallback(_callback,x);
798 if (_shutdownOut)
799 getEndPoint().shutdownOutput();
800 }
801
802 @Override
803 public String toString()
804 {
805 return String.format("%s[i=%s,cb=%s]",super.toString(),_info,_callback);
806 }
807 }
808 }