View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
4   //  ------------------------------------------------------------------------
5   //  All rights reserved. This program and the accompanying materials
6   //  are made available under the terms of the Eclipse Public License v1.0
7   //  and Apache License v2.0 which accompanies this distribution.
8   //
9   //      The Eclipse Public License is available at
10  //      http://www.eclipse.org/legal/epl-v10.html
11  //
12  //      The Apache License v2.0 is available at
13  //      http://www.opensource.org/licenses/apache2.0.php
14  //
15  //  You may elect to redistribute this code under either of these licenses.
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  
26  import org.eclipse.jetty.http.HttpGenerator;
27  import org.eclipse.jetty.http.HttpGenerator.ResponseInfo;
28  import org.eclipse.jetty.http.HttpHeader;
29  import org.eclipse.jetty.http.HttpHeaderValue;
30  import org.eclipse.jetty.http.HttpMethod;
31  import org.eclipse.jetty.http.HttpParser;
32  import org.eclipse.jetty.http.HttpStatus;
33  import org.eclipse.jetty.http.HttpVersion;
34  import org.eclipse.jetty.io.AbstractConnection;
35  import org.eclipse.jetty.io.ByteBufferPool;
36  import org.eclipse.jetty.io.Connection;
37  import org.eclipse.jetty.io.EndPoint;
38  import org.eclipse.jetty.io.EofException;
39  import org.eclipse.jetty.util.BufferUtil;
40  import org.eclipse.jetty.util.Callback;
41  import org.eclipse.jetty.util.IteratingCallback;
42  import org.eclipse.jetty.util.log.Log;
43  import org.eclipse.jetty.util.log.Logger;
44  
45  /**
46   * <p>A {@link Connection} that handles the HTTP protocol.</p>
47   */
48  public class HttpConnection extends AbstractConnection implements Runnable, HttpTransport, Connection.UpgradeFrom
49  {
50      public static final String UPGRADE_CONNECTION_ATTRIBUTE = "org.eclipse.jetty.server.HttpConnection.UPGRADE";
51      private static final boolean REQUEST_BUFFER_DIRECT=false;
52      private static final boolean HEADER_BUFFER_DIRECT=false;
53      private static final boolean CHUNK_BUFFER_DIRECT=false;
54      private static final Logger LOG = Log.getLogger(HttpConnection.class);
55      private static final ThreadLocal<HttpConnection> __currentConnection = new ThreadLocal<>();
56  
57      private final HttpConfiguration _config;
58      private final Connector _connector;
59      private final ByteBufferPool _bufferPool;
60      private final HttpGenerator _generator;
61      private final HttpChannelOverHttp _channel;
62      private final HttpParser _parser;
63      private volatile ByteBuffer _requestBuffer = null;
64      private volatile ByteBuffer _chunk = null;
65      private final SendCallback _sendCallback = new SendCallback();
66  
67  
68      /* ------------------------------------------------------------ */
69      /** Get the current connection that this thread is dispatched to.
70       * Note that a thread may be processing a request asynchronously and 
71       * thus not be dispatched to the connection.  
72       * @see Request#getAttribute(String) for a more general way to access the HttpConnection
73       * @return the current HttpConnection or null
74       */
75      public static HttpConnection getCurrentConnection()
76      {
77          return __currentConnection.get();
78      }
79  
80      protected static HttpConnection setCurrentConnection(HttpConnection connection)
81      {
82          HttpConnection last=__currentConnection.get();
83          __currentConnection.set(connection);
84          return last;
85      }
86  
87      public HttpConfiguration getHttpConfiguration()
88      {
89          return _config;
90      }
91  
92      public HttpConnection(HttpConfiguration config, Connector connector, EndPoint endPoint)
93      {
94          // Tell AbstractConnector executeOnFillable==true because we want the same thread that
95          // does the HTTP parsing to handle the request so its cache is hot
96          super(endPoint, connector.getExecutor(),true);
97  
98          _config = config;
99          _connector = connector;
100         _bufferPool = _connector.getByteBufferPool();
101         _generator = newHttpGenerator();
102         HttpInput<ByteBuffer> input = newHttpInput();
103         _channel = newHttpChannel(input);
104         _parser = newHttpParser();
105         if (LOG.isDebugEnabled())
106             LOG.debug("New HTTP Connection {}", this);
107     }
108 
109     protected HttpGenerator newHttpGenerator()
110     {
111         return new HttpGenerator(_config.getSendServerVersion(),_config.getSendXPoweredBy());
112     }
113     
114     protected HttpInput<ByteBuffer> newHttpInput()
115     {
116         return new HttpInputOverHTTP(this);
117     }
118     
119     protected HttpChannelOverHttp newHttpChannel(HttpInput<ByteBuffer> httpInput)
120     {
121         return new HttpChannelOverHttp(_connector, _config, getEndPoint(), this, httpInput);
122     }
123     
124     protected HttpParser newHttpParser()
125     {
126         return new HttpParser(newRequestHandler(), getHttpConfiguration().getRequestHeaderSize());
127     }
128 
129     protected HttpParser.RequestHandler<ByteBuffer> newRequestHandler()
130     {
131         return _channel;
132     }
133 
134     public Server getServer()
135     {
136         return _connector.getServer();
137     }
138 
139     public Connector getConnector()
140     {
141         return _connector;
142     }
143 
144     public HttpChannel<?> getHttpChannel()
145     {
146         return _channel;
147     }
148 
149     public HttpParser getParser()
150     {
151         return _parser;
152     }
153 
154     @Override
155     public int getMessagesIn()
156     {
157         return getHttpChannel().getRequests();
158     }
159 
160     @Override
161     public int getMessagesOut()
162     {
163         return getHttpChannel().getRequests();
164     }
165 
166     @Override
167     public ByteBuffer onUpgradeFrom()
168     {
169         if (BufferUtil.hasContent(_requestBuffer))
170         {
171             ByteBuffer buffer = _requestBuffer;
172             _requestBuffer=null;
173             return buffer;
174         }
175         return null;
176     }
177 
178     void releaseRequestBuffer()
179     {
180         if (_requestBuffer != null && !_requestBuffer.hasRemaining())
181         {
182             ByteBuffer buffer=_requestBuffer;
183             _requestBuffer=null;
184             _bufferPool.release(buffer);
185         }
186     }
187     
188     public ByteBuffer getRequestBuffer()
189     {
190         if (_requestBuffer == null)
191             _requestBuffer = _bufferPool.acquire(getInputBufferSize(), REQUEST_BUFFER_DIRECT);
192         return _requestBuffer;
193     }
194 
195     /**
196      * <p>Parses and handles HTTP messages.</p>
197      * <p>This method is called when this {@link Connection} is ready to read bytes from the {@link EndPoint}.
198      * However, it can also be called if there is unconsumed data in the _requestBuffer, as a result of
199      * resuming a suspended request when there is a pipelined request already read into the buffer.</p>
200      * <p>This method fills bytes and parses them until either: EOF is filled; 0 bytes are filled;
201      * the HttpChannel finishes handling; or the connection has changed.</p>
202      */
203     @Override
204     public void onFillable()
205     {
206         if (LOG.isDebugEnabled())
207             LOG.debug("{} onFillable {}", this, _channel.getState());
208 
209         final HttpConnection last=setCurrentConnection(this);
210         int filled=Integer.MAX_VALUE;
211         boolean suspended=false;
212         try
213         {
214             // while not suspended and not upgraded
215             while (!suspended && getEndPoint().getConnection()==this)
216             {
217                 // Do we need some data to parse
218                 if (BufferUtil.isEmpty(_requestBuffer))
219                 {
220                     // If the previous iteration filled 0 bytes or saw a close, then break here 
221                     if (filled<=0)
222                         break;
223                         
224                     // Can we fill?
225                     if(getEndPoint().isInputShutdown())
226                     {
227                         // No pretend we read -1
228                         filled=-1;
229                         _parser.atEOF();
230                     }
231                     else
232                     {
233                         // Get a buffer
234                         // We are not in a race here for the request buffer as we have not yet received a request,
235                         // so there are not an possible legal threads calling #parseContent or #completed.
236                         _requestBuffer = getRequestBuffer();
237 
238                         // fill
239                         filled = getEndPoint().fill(_requestBuffer);
240                         if (filled==0) // Do a retry on fill 0 (optimization for SSL connections)
241                             filled = getEndPoint().fill(_requestBuffer);
242                         
243                         // tell parser
244                         if (filled < 0)
245                             _parser.atEOF();
246                     }
247                 }
248                 
249                 // Parse the buffer
250                 if (_parser.parseNext(_requestBuffer==null?BufferUtil.EMPTY_BUFFER:_requestBuffer))
251                 {
252                     // The parser returned true, which indicates the channel is ready to handle a request.
253                     // Call the channel and this will either handle the request/response to completion OR,
254                     // if the request suspends, the request/response will be incomplete so the outer loop will exit.
255                     // Not that onFillable no longer manipulates the request buffer from this point and that is
256                     // left to threads calling #completed or #parseContent (which may be this thread inside handle())
257                     suspended = !_channel.handle();
258                 }
259                 else
260                 {
261                     // We parsed what we could, recycle the request buffer
262                     // We are not in a race here for the request buffer as we have not yet received a request,
263                     // so there are not an possible legal threads calling #parseContent or #completed.
264                     releaseRequestBuffer();
265                 }
266             }
267         }
268         catch (EofException e)
269         {
270             LOG.debug(e);
271         }
272         catch (Exception e)
273         {
274             if (_parser.isIdle())
275                 LOG.debug(e);
276             else
277                 LOG.warn(this.toString(), e);
278             close();
279         }
280         finally
281         {                        
282             setCurrentConnection(last);
283             if (!suspended && getEndPoint().isOpen() && getEndPoint().getConnection()==this)
284             {
285                 fillInterested();
286             }
287         }
288     }
289 
290     /* ------------------------------------------------------------ */
291     /** Fill and parse data looking for content
292      * @throws IOException
293      */
294     protected void parseContent() throws IOException
295     {
296         // Not in a race here for the request buffer with #onFillable because an async consumer of
297         // content would only be started after onFillable has given up control.
298         // In a little bit of a race with #completed, but then not sure if it is legal to be doing 
299         // async calls to IO and have a completed call at the same time.
300         ByteBuffer requestBuffer = getRequestBuffer();
301 
302         while (_parser.inContentState())
303         {
304             // Can the parser progress (even with an empty buffer)
305             boolean parsed = _parser.parseNext(requestBuffer==null?BufferUtil.EMPTY_BUFFER:requestBuffer);
306 
307             // No, we can we try reading some content?
308             if (BufferUtil.isEmpty(requestBuffer) && getEndPoint().isInputShutdown())
309             {
310                 _parser.atEOF();
311                 if (parsed)
312                     break;
313                 continue;
314             }
315 
316             if (parsed)
317                 break;
318             
319             // OK lets read some data
320             int filled=getEndPoint().fill(requestBuffer);
321             if (LOG.isDebugEnabled()) // Avoid boxing of variable 'filled'
322                 LOG.debug("{} filled {}",this,filled);
323             if (filled<=0)
324             {
325                 if (filled<0)
326                 {
327                     _parser.atEOF();
328                     continue;
329                 }
330                 break;
331             }
332         }
333     }
334     
335     @Override
336     public void completed()
337     {
338         // Handle connection upgrades
339         if (_channel.getResponse().getStatus() == HttpStatus.SWITCHING_PROTOCOLS_101)
340         {
341             Connection connection = (Connection)_channel.getRequest().getAttribute(UPGRADE_CONNECTION_ATTRIBUTE);
342             if (connection != null)
343             {
344                 _channel.getState().upgrade();
345                 getEndPoint().upgrade(connection);
346                 _channel.reset();
347                 _parser.reset();
348                 _generator.reset();
349                 releaseRequestBuffer();
350                 return;
351             }
352         }
353         
354         // Finish consuming the request
355         // If we are still expecting
356         if (_channel.isExpecting100Continue())
357         {
358             // close to seek EOF
359             _parser.close();
360         }
361         else if (_parser.inContentState() && _generator.isPersistent())
362         {
363             // If we are async, then we have problems to complete neatly
364             if (_channel.getRequest().getHttpInput().isAsync())
365             {
366                 if (LOG.isDebugEnabled())
367                     LOG.debug("unconsumed async input {}", this);
368                 _channel.abort();
369             }
370             else
371             {
372                 if (LOG.isDebugEnabled())
373                     LOG.debug("unconsumed input {}", this);
374                 // Complete reading the request
375                 if (!_channel.getRequest().getHttpInput().consumeAll())
376                     _channel.abort();
377             }
378         }
379 
380         // Reset the channel, parsers and generator
381         _channel.reset();
382         if (_generator.isPersistent() && !_parser.isClosed())
383             _parser.reset();
384         else
385             _parser.close();
386         
387         // Not in a race here with onFillable, because it has given up control before calling handle.
388         // in a slight race with #completed, but not sure what to do with that anyway.
389         releaseRequestBuffer();
390         if (_chunk!=null)
391             _bufferPool.release(_chunk);
392         _chunk=null;
393         _generator.reset();
394 
395         // if we are not called from the onfillable thread, schedule completion
396         if (getCurrentConnection()!=this)
397         {
398             // If we are looking for the next request
399             if (_parser.isStart())
400             {
401                 // if the buffer is empty
402                 if (BufferUtil.isEmpty(_requestBuffer))
403                 {
404                     // look for more data
405                     fillInterested();
406                 }
407                 // else if we are still running
408                 else if (getConnector().isRunning())
409                 {
410                     // Dispatched to handle a pipelined request
411                     try
412                     {
413                         getExecutor().execute(this);
414                     }
415                     catch (RejectedExecutionException e)
416                     {
417                         if (getConnector().isRunning())
418                             LOG.warn(e);
419                         else
420                             LOG.ignore(e);
421                         getEndPoint().close();
422                     }
423                 }
424                 else
425                 {
426                     getEndPoint().close();
427                 }
428             }
429             // else the parser must be closed, so seek the EOF if we are still open 
430             else if (getEndPoint().isOpen())
431                 fillInterested();
432         }
433     }
434 
435     @Override
436     protected void onFillInterestedFailed(Throwable cause)
437     {
438         _parser.close();
439         super.onFillInterestedFailed(cause);
440     }
441 
442     @Override
443     public void onOpen()
444     {
445         super.onOpen();
446         fillInterested();
447     }
448 
449     @Override
450     public void onClose()
451     {
452         _sendCallback.close();
453         super.onClose();
454     }
455 
456     @Override
457     public void run()
458     {
459         onFillable();
460     }
461 
462     @Override
463     public void send(ResponseInfo info, ByteBuffer content, boolean lastContent, Callback callback)
464     {
465         // If we are still expecting a 100 continues when we commit
466         if (info!=null && _channel.isExpecting100Continue())
467             // then we can't be persistent
468             _generator.setPersistent(false);
469         
470         if(_sendCallback.reset(info,content,lastContent,callback))
471             _sendCallback.iterate();
472     }
473 
474     @Override
475     public void send(ByteBuffer content, boolean lastContent, Callback callback)
476     {
477         if (!lastContent && BufferUtil.isEmpty(content))
478             callback.succeeded();
479         else if (_sendCallback.reset(null,content,lastContent,callback))
480             _sendCallback.iterate();
481     }
482     
483     protected class HttpChannelOverHttp extends HttpChannel<ByteBuffer>
484     {        
485         public HttpChannelOverHttp(Connector connector, HttpConfiguration config, EndPoint endPoint, HttpTransport transport, HttpInput<ByteBuffer> input)
486         {
487             super(connector,config,endPoint,transport,input);
488         }
489         
490         @Override
491         public void earlyEOF()
492         {
493             // If we have no request yet, just close
494             if (getRequest().getMethod()==null)
495                 close();
496             else
497                 super.earlyEOF();
498         }
499 
500         @Override
501         public boolean content(ByteBuffer item)
502         {
503             super.content(item);
504             return true;
505         }
506 
507         @Override
508         public void badMessage(int status, String reason)
509         {
510             _generator.setPersistent(false);
511             super.badMessage(status,reason);
512         }
513 
514         @Override
515         public boolean headerComplete()
516         {
517             boolean persistent;
518             HttpVersion version = getHttpVersion();
519 
520             switch (version)
521             {
522                 case HTTP_0_9:
523                 {
524                     persistent = false;
525                     break;
526                 }
527                 case HTTP_1_0:
528                 {
529                     persistent = getRequest().getHttpFields().contains(HttpHeader.CONNECTION, HttpHeaderValue.KEEP_ALIVE.asString());
530                     if (!persistent)
531                         persistent = HttpMethod.CONNECT.is(getRequest().getMethod());
532                     if (persistent)
533                         getResponse().getHttpFields().add(HttpHeader.CONNECTION, HttpHeaderValue.KEEP_ALIVE);
534                     break;
535                 }
536                 case HTTP_1_1:
537                 {
538                     persistent = !getRequest().getHttpFields().contains(HttpHeader.CONNECTION, HttpHeaderValue.CLOSE.asString());
539                     if (!persistent)
540                         persistent = HttpMethod.CONNECT.is(getRequest().getMethod());
541                     if (!persistent)
542                         getResponse().getHttpFields().add(HttpHeader.CONNECTION, HttpHeaderValue.CLOSE);
543                     break;
544                 }
545                 case HTTP_2:
546                 {
547                     persistent=false;
548                     badMessage(400,null);
549                     return true;
550                 }
551                 default:
552                 {
553                     throw new IllegalStateException();
554                 }
555             }
556 
557             if (!persistent)
558                 _generator.setPersistent(false);
559 
560             if (!super.headerComplete())
561                 return false;
562             
563             // Should we delay dispatch until we have some content?
564             // We should not delay if there is no content expect or client is expecting 100 or the response is already committed or the request buffer already has something in it to parse
565             if (getHttpConfiguration().isDelayDispatchUntilContent() && _parser.getContentLength() > 0 &&
566                     !isExpecting100Continue() && !isCommitted() && BufferUtil.isEmpty(_requestBuffer))
567                 return false;
568 
569             return true;
570         }
571 
572         @Override
573         protected void handleException(Throwable x)
574         {
575             _generator.setPersistent(false);
576             super.handleException(x);
577         }
578 
579         @Override
580         public void abort()
581         {
582             super.abort();
583             _generator.setPersistent(false);
584         }
585 
586         @Override
587         public boolean messageComplete()
588         {
589             super.messageComplete();
590             return false;
591         }
592     }
593 
594     private class SendCallback extends IteratingCallback
595     {
596         private ResponseInfo _info;
597         private ByteBuffer _content;
598         private boolean _lastContent;
599         private Callback _callback;
600         private ByteBuffer _header;
601         private boolean _shutdownOut;
602 
603         private SendCallback()
604         {
605             super(true);
606         }
607 
608         private boolean reset(ResponseInfo info, ByteBuffer content, boolean last, Callback callback)
609         {
610             if (reset())
611             {
612                 _info = info;
613                 _content = content;
614                 _lastContent = last;
615                 _callback = callback;
616                 _header = null;
617                 _shutdownOut = false;
618                 return true;
619             }
620             
621             if (isClosed())
622                 callback.failed(new EofException());
623             else
624                 callback.failed(new WritePendingException());
625             return false;
626         }
627 
628         @Override
629         public Action process() throws Exception
630         {
631             if (_callback==null)
632                 throw new IllegalStateException();
633             
634             ByteBuffer chunk = _chunk;
635             while (true)
636             {
637                 HttpGenerator.Result result = _generator.generateResponse(_info, _header, chunk, _content, _lastContent);
638                 if (LOG.isDebugEnabled())
639                     LOG.debug("{} generate: {} ({},{},{})@{}",
640                         this,
641                         result,
642                         BufferUtil.toSummaryString(_header),
643                         BufferUtil.toSummaryString(_content),
644                         _lastContent,
645                         _generator.getState());
646 
647                 switch (result)
648                 {
649                     case NEED_HEADER:
650                     {
651                         _header = _bufferPool.acquire(_config.getResponseHeaderSize(), HEADER_BUFFER_DIRECT);    
652                         continue;
653                     }
654                     case NEED_CHUNK:
655                     {
656                         chunk = _chunk = _bufferPool.acquire(HttpGenerator.CHUNK_SIZE, CHUNK_BUFFER_DIRECT);
657                         continue;
658                     }
659                     case FLUSH:
660                     {
661                         // Don't write the chunk or the content if this is a HEAD response, or any other type of response that should have no content
662                         if (_channel.getRequest().isHead() || _generator.isNoContent())
663                         {
664                             BufferUtil.clear(chunk);
665                             BufferUtil.clear(_content);
666                         }
667 
668                         // If we have a header
669                         if (BufferUtil.hasContent(_header))
670                         {
671                             if (BufferUtil.hasContent(_content))
672                             {
673                                 if (BufferUtil.hasContent(chunk))
674                                     getEndPoint().write(this, _header, chunk, _content);
675                                 else
676                                     getEndPoint().write(this, _header, _content);
677                             }
678                             else
679                                 getEndPoint().write(this, _header);
680                         }
681                         else if (BufferUtil.hasContent(chunk))
682                         {
683                             if (BufferUtil.hasContent(_content))
684                                 getEndPoint().write(this, chunk, _content);
685                             else
686                                 getEndPoint().write(this, chunk);
687                         }
688                         else if (BufferUtil.hasContent(_content))
689                         {
690                             getEndPoint().write(this, _content);
691                         }
692                         else
693                         {
694                             succeeded(); // nothing to write
695                         }
696                         return Action.SCHEDULED;
697                     }
698                     case SHUTDOWN_OUT:
699                     {
700                         _shutdownOut=true;
701                         continue;
702                     }
703                     case DONE:
704                     {
705                         return Action.SUCCEEDED;
706                     }
707                     case CONTINUE:
708                     {
709                         break;
710                     }
711                     default:
712                     {
713                         throw new IllegalStateException("generateResponse="+result);
714                     }
715                 }
716             }
717         }
718 
719         private void releaseHeader()
720         {
721             ByteBuffer h=_header;
722             _header=null;
723             if (h!=null)
724                 _bufferPool.release(h);
725         }
726         
727         @Override
728         protected void onCompleteSuccess()
729         {
730             releaseHeader();
731             _callback.succeeded();
732             if (_shutdownOut)
733                 getEndPoint().shutdownOutput();
734         }
735 
736         @Override
737         public void onCompleteFailure(final Throwable x)
738         {
739             releaseHeader();
740             failedCallback(_callback,x);
741             if (_shutdownOut)
742                 getEndPoint().shutdownOutput();
743         }
744         
745         @Override
746         public String toString()
747         {
748             return String.format("%s[i=%s,cb=%s]",super.toString(),_info,_callback);
749         }
750     }
751 
752 
753     @Override
754     public void abort()
755     {
756         // Do a direct close of the output, as this may indicate to a client that the 
757         // response is bad either with RST or by abnormal completion of chunked response.
758         getEndPoint().close();
759     }
760 
761 }