2 // ========================================================================
3 // Copyright (c) 1995-2014 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.
9 // The Eclipse Public License is available at
10 // http://www.eclipse.org/legal/epl-v10.html
12 // The Apache License v2.0 is available at
13 // http://www.opensource.org/licenses/apache2.0.php
15 // You may elect to redistribute this code under either of these licenses.
16 // ========================================================================
19 package org.eclipse.jetty.server;
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.util.concurrent.RejectedExecutionException;
25 import org.eclipse.jetty.http.HttpGenerator;
26 import org.eclipse.jetty.http.HttpGenerator.ResponseInfo;
27 import org.eclipse.jetty.http.HttpHeader;
28 import org.eclipse.jetty.http.HttpHeaderValue;
29 import org.eclipse.jetty.http.HttpMethod;
30 import org.eclipse.jetty.http.HttpParser;
31 import org.eclipse.jetty.http.HttpStatus;
32 import org.eclipse.jetty.http.HttpVersion;
33 import org.eclipse.jetty.io.AbstractConnection;
34 import org.eclipse.jetty.io.ByteBufferPool;
35 import org.eclipse.jetty.io.Connection;
36 import org.eclipse.jetty.io.EndPoint;
37 import org.eclipse.jetty.io.EofException;
38 import org.eclipse.jetty.util.BufferUtil;
39 import org.eclipse.jetty.util.Callback;
40 import org.eclipse.jetty.util.IteratingCallback;
41 import org.eclipse.jetty.util.log.Log;
42 import org.eclipse.jetty.util.log.Logger;
45 * <p>A {@link Connection} that handles the HTTP protocol.</p>
47 public class HttpConnection extends AbstractConnection implements Runnable, HttpTransport
49 public static final String UPGRADE_CONNECTION_ATTRIBUTE = "org.eclipse.jetty.server.HttpConnection.UPGRADE";
50 private static final boolean REQUEST_BUFFER_DIRECT=false;
51 private static final boolean HEADER_BUFFER_DIRECT=false;
52 private static final boolean CHUNK_BUFFER_DIRECT=false;
53 private static final Logger LOG = Log.getLogger(HttpConnection.class);
54 private static final ThreadLocal<HttpConnection> __currentConnection = new ThreadLocal<>();
56 private final HttpConfiguration _config;
57 private final Connector _connector;
58 private final ByteBufferPool _bufferPool;
59 private final HttpGenerator _generator;
60 private final HttpChannelOverHttp _channel;
61 private final HttpParser _parser;
62 private volatile ByteBuffer _requestBuffer = null;
63 private volatile ByteBuffer _chunk = null;
66 /* ------------------------------------------------------------ */
67 /** Get the current connection that this thread is dispatched to.
68 * Note that a thread may be processing a request asynchronously and
69 * thus not be dispatched to the connection.
70 * @see Request#getAttribute(String) for a more general way to access the HttpConnection
71 * @return the current HttpConnection or null
73 public static HttpConnection getCurrentConnection()
75 return __currentConnection.get();
78 protected static HttpConnection setCurrentConnection(HttpConnection connection)
80 HttpConnection last=__currentConnection.get();
82 __currentConnection.remove();
84 __currentConnection.set(connection);
88 public HttpConfiguration getHttpConfiguration()
93 public HttpConnection(HttpConfiguration config, Connector connector, EndPoint endPoint)
95 // Tell AbstractConnector executeOnFillable==true because we want the same thread that
96 // does the HTTP parsing to handle the request so its cache is hot
97 super(endPoint, connector.getExecutor(),true);
100 _connector = connector;
101 _bufferPool = _connector.getByteBufferPool();
102 _generator = newHttpGenerator();
103 HttpInput<ByteBuffer> input = newHttpInput();
104 _channel = newHttpChannel(input);
105 _parser = newHttpParser();
106 LOG.debug("New HTTP Connection {}", this);
109 protected HttpGenerator newHttpGenerator()
111 return new HttpGenerator(_config.getSendServerVersion(),_config.getSendXPoweredBy());
114 protected HttpInput<ByteBuffer> newHttpInput()
116 return new HttpInputOverHTTP(this);
119 protected HttpChannelOverHttp newHttpChannel(HttpInput<ByteBuffer> httpInput)
121 return new HttpChannelOverHttp(_connector, _config, getEndPoint(), this, httpInput);
124 protected HttpParser newHttpParser()
126 return new HttpParser(newRequestHandler(), getHttpConfiguration().getRequestHeaderSize());
129 protected HttpParser.RequestHandler<ByteBuffer> newRequestHandler()
134 public Server getServer()
136 return _connector.getServer();
139 public Connector getConnector()
144 public HttpChannel<?> getHttpChannel()
149 public HttpParser getParser()
155 public int getMessagesIn()
157 return getHttpChannel().getRequests();
161 public int getMessagesOut()
163 return getHttpChannel().getRequests();
166 void releaseRequestBuffer()
168 if (_requestBuffer != null && !_requestBuffer.hasRemaining())
170 ByteBuffer buffer=_requestBuffer;
172 _bufferPool.release(buffer);
176 public ByteBuffer getRequestBuffer()
178 if (_requestBuffer == null)
179 _requestBuffer = _bufferPool.acquire(getInputBufferSize(), REQUEST_BUFFER_DIRECT);
180 return _requestBuffer;
184 * <p>Parses and handles HTTP messages.</p>
185 * <p>This method is called when this {@link Connection} is ready to read bytes from the {@link EndPoint}.
186 * However, it can also be called if there is unconsumed data in the _requestBuffer, as a result of
187 * resuming a suspended request when there is a pipelined request already read into the buffer.</p>
188 * <p>This method fills bytes and parses them until either: EOF is filled; 0 bytes are filled;
189 * the HttpChannel finishes handling; or the connection has changed.</p>
192 public void onFillable()
194 LOG.debug("{} onFillable {}", this, _channel.getState());
196 final HttpConnection last=setCurrentConnection(this);
197 int filled=Integer.MAX_VALUE;
198 boolean suspended=false;
201 // while not suspended and not upgraded
202 while (!suspended && getEndPoint().getConnection()==this)
204 // Do we need some data to parse
205 if (BufferUtil.isEmpty(_requestBuffer))
207 // If the previous iteration filled 0 bytes or saw a close, then break here
212 if(getEndPoint().isInputShutdown())
214 // No pretend we read -1
221 // We are not in a race here for the request buffer as we have not yet received a request,
222 // so there are not an possible legal threads calling #parseContent or #completed.
223 _requestBuffer = getRequestBuffer();
226 filled = getEndPoint().fill(_requestBuffer);
227 if (filled==0) // Do a retry on fill 0 (optimization for SSL connections)
228 filled = getEndPoint().fill(_requestBuffer);
237 if (_parser.parseNext(_requestBuffer==null?BufferUtil.EMPTY_BUFFER:_requestBuffer))
239 // The parser returned true, which indicates the channel is ready to handle a request.
240 // Call the channel and this will either handle the request/response to completion OR,
241 // if the request suspends, the request/response will be incomplete so the outer loop will exit.
242 // Not that onFillable no longer manipulates the request buffer from this point and that is
243 // left to threads calling #completed or #parseContent (which may be this thread inside handle())
244 suspended = !_channel.handle();
248 // We parsed what we could, recycle the request buffer
249 // We are not in a race here for the request buffer as we have not yet received a request,
250 // so there are not an possible legal threads calling #parseContent or #completed.
251 releaseRequestBuffer();
255 catch (EofException e)
261 if (_parser.isIdle())
264 LOG.warn(this.toString(), e);
269 setCurrentConnection(last);
270 if (!suspended && getEndPoint().isOpen() && getEndPoint().getConnection()==this)
277 /* ------------------------------------------------------------ */
278 /** Fill and parse data looking for content
279 * @throws IOException
281 protected void parseContent() throws IOException
283 // Not in a race here for the request buffer with #onFillable because an async consumer of
284 // content would only be started after onFillable has given up control.
285 // In a little bit of a race with #completed, but then not sure if it is legal to be doing
286 // async calls to IO and have a completed call at the same time.
287 ByteBuffer requestBuffer = getRequestBuffer();
289 while (_parser.inContentState())
291 // Can the parser progress (even with an empty buffer)
292 boolean parsed = _parser.parseNext(requestBuffer==null?BufferUtil.EMPTY_BUFFER:requestBuffer);
294 // No, we can we try reading some content?
295 if (BufferUtil.isEmpty(requestBuffer) && getEndPoint().isInputShutdown())
306 // OK lets read some data
307 int filled=getEndPoint().fill(requestBuffer);
308 if (LOG.isDebugEnabled()) // Avoid boxing of variable 'filled'
309 LOG.debug("{} filled {}",this,filled);
323 public void completed()
325 // Handle connection upgrades
326 if (_channel.getResponse().getStatus() == HttpStatus.SWITCHING_PROTOCOLS_101)
328 Connection connection = (Connection)_channel.getRequest().getAttribute(UPGRADE_CONNECTION_ATTRIBUTE);
329 if (connection != null)
331 LOG.debug("Upgrade from {} to {}", this, connection);
333 getEndPoint().setConnection(connection);
338 releaseRequestBuffer();
343 // Finish consuming the request
344 // If we are still expecting
345 if (_channel.isExpecting100Continue())
348 else if (_parser.inContentState() && _generator.isPersistent())
349 // Complete reading the request
350 _channel.getRequest().getHttpInput().consumeAll();
352 // Reset the channel, parsers and generator
354 if (_generator.isPersistent() && !_parser.isClosed())
359 // Not in a race here with onFillable, because it has given up control before calling handle.
360 // in a slight race with #completed, but not sure what to do with that anyway.
361 releaseRequestBuffer();
363 _bufferPool.release(_chunk);
367 // if we are not called from the onfillable thread, schedule completion
368 if (getCurrentConnection()!=this)
370 // If we are looking for the next request
371 if (_parser.isStart())
373 // if the buffer is empty
374 if (BufferUtil.isEmpty(_requestBuffer))
376 // look for more data
379 // else if we are still running
380 else if (getConnector().isRunning())
382 // Dispatched to handle a pipelined request
385 getExecutor().execute(this);
387 catch (RejectedExecutionException e)
389 if (getConnector().isRunning())
393 getEndPoint().close();
398 getEndPoint().close();
401 // else the parser must be closed, so seek the EOF if we are still open
402 else if (getEndPoint().isOpen())
408 protected void onFillInterestedFailed(Throwable cause)
411 super.onFillInterestedFailed(cause);
429 public void send(ResponseInfo info, ByteBuffer content, boolean lastContent, Callback callback)
432 new ContentCallback(content,lastContent,callback).iterate();
435 // If we are still expecting a 100 continues
436 if (_channel.isExpecting100Continue())
437 // then we can't be persistent
438 _generator.setPersistent(false);
439 new CommitCallback(info,content,lastContent,callback).iterate();
444 public void send(ByteBuffer content, boolean lastContent, Callback callback)
446 new ContentCallback(content,lastContent,callback).iterate();
450 protected class HttpChannelOverHttp extends HttpChannel<ByteBuffer>
452 public HttpChannelOverHttp(Connector connector, HttpConfiguration config, EndPoint endPoint, HttpTransport transport, HttpInput<ByteBuffer> input)
454 super(connector,config,endPoint,transport,input);
458 public void earlyEOF()
460 // If we have no request yet, just close
461 if (getRequest().getMethod()==null)
468 public boolean content(ByteBuffer item)
475 public void badMessage(int status, String reason)
477 _generator.setPersistent(false);
478 super.badMessage(status,reason);
482 public boolean headerComplete()
485 HttpVersion version = getHttpVersion();
496 persistent = getRequest().getHttpFields().contains(HttpHeader.CONNECTION, HttpHeaderValue.KEEP_ALIVE.asString());
498 persistent = HttpMethod.CONNECT.is(getRequest().getMethod());
500 getResponse().getHttpFields().add(HttpHeader.CONNECTION, HttpHeaderValue.KEEP_ALIVE);
505 persistent = !getRequest().getHttpFields().contains(HttpHeader.CONNECTION, HttpHeaderValue.CLOSE.asString());
507 persistent = HttpMethod.CONNECT.is(getRequest().getMethod());
509 getResponse().getHttpFields().add(HttpHeader.CONNECTION, HttpHeaderValue.CLOSE);
514 throw new IllegalStateException();
519 _generator.setPersistent(false);
521 return super.headerComplete();
525 protected void handleException(Throwable x)
527 _generator.setPersistent(false);
528 super.handleException(x);
534 getEndPoint().shutdownOutput();
539 public boolean messageComplete()
541 super.messageComplete();
546 private class CommitCallback extends IteratingCallback
548 final ByteBuffer _content;
549 final boolean _lastContent;
550 final ResponseInfo _info;
551 final Callback _callback;
554 CommitCallback(ResponseInfo info, ByteBuffer content, boolean last, Callback callback)
563 public Action process() throws Exception
565 ByteBuffer chunk = _chunk;
568 HttpGenerator.Result result = _generator.generateResponse(_info, _header, chunk, _content, _lastContent);
569 if (LOG.isDebugEnabled())
570 LOG.debug("{} generate: {} ({},{},{})@{}",
573 BufferUtil.toSummaryString(_header),
574 BufferUtil.toSummaryString(_content),
576 _generator.getState());
582 // Look for optimisation to avoid allocating a _header buffer
584 Cannot use this optimisation unless we work out how not to overwrite data in user passed arrays.
585 if (_lastContent && _content!=null && !_content.isReadOnly() && _content.hasArray() && BufferUtil.space(_content)>_config.getResponseHeaderSize() )
587 // use spare space in content buffer for header buffer
588 int p=_content.position();
589 int l=_content.limit();
590 _content.position(l);
591 _content.limit(l+_config.getResponseHeaderSize());
592 _header=_content.slice();
594 _content.position(p);
599 _header = _bufferPool.acquire(_config.getResponseHeaderSize(), HEADER_BUFFER_DIRECT);
605 chunk = _chunk = _bufferPool.acquire(HttpGenerator.CHUNK_SIZE, CHUNK_BUFFER_DIRECT);
610 // Don't write the chunk or the content if this is a HEAD response
611 if (_channel.getRequest().isHead())
613 BufferUtil.clear(chunk);
614 BufferUtil.clear(_content);
617 // If we have a header
618 if (BufferUtil.hasContent(_header))
620 if (BufferUtil.hasContent(_content))
622 if (BufferUtil.hasContent(chunk))
623 getEndPoint().write(this, _header, chunk, _content);
625 getEndPoint().write(this, _header, _content);
628 getEndPoint().write(this, _header);
630 else if (BufferUtil.hasContent(chunk))
632 if (BufferUtil.hasContent(_content))
633 getEndPoint().write(this, chunk, _content);
635 getEndPoint().write(this, chunk);
637 else if (BufferUtil.hasContent(_content))
639 getEndPoint().write(this, _content);
643 return Action.SCHEDULED;
647 getEndPoint().shutdownOutput();
654 // don't release header in spare content buffer
655 if (!_lastContent || _content==null || !_content.hasArray() || !_header.hasArray() || _content.array()!=_header.array())
656 _bufferPool.release(_header);
658 return Action.SUCCEEDED;
666 throw new IllegalStateException("generateResponse="+result);
673 protected void completed()
675 _callback.succeeded();
679 public void failed(final Throwable x)
682 failedCallback(_callback,x);
686 private class ContentCallback extends IteratingCallback
688 final ByteBuffer _content;
689 final boolean _lastContent;
690 final Callback _callback;
692 ContentCallback(ByteBuffer content, boolean last, Callback callback)
700 public Action process() throws Exception
702 ByteBuffer chunk = _chunk;
705 HttpGenerator.Result result = _generator.generateResponse(null, null, chunk, _content, _lastContent);
706 if (LOG.isDebugEnabled())
707 LOG.debug("{} generate: {} ({},{})@{}",
710 BufferUtil.toSummaryString(_content),
712 _generator.getState());
717 throw new IllegalStateException();
720 chunk = _chunk = _bufferPool.acquire(HttpGenerator.CHUNK_SIZE, CHUNK_BUFFER_DIRECT);
725 // Don't write the chunk or the content if this is a HEAD response
726 if (_channel.getRequest().isHead())
728 BufferUtil.clear(chunk);
729 BufferUtil.clear(_content);
732 else if (BufferUtil.hasContent(chunk))
734 if (BufferUtil.hasContent(_content))
735 getEndPoint().write(this, chunk, _content);
737 getEndPoint().write(this, chunk);
739 else if (BufferUtil.hasContent(_content))
741 getEndPoint().write(this, _content);
745 return Action.SCHEDULED;
749 getEndPoint().shutdownOutput();
754 return Action.SUCCEEDED;
762 throw new IllegalStateException("generateResponse="+result);
769 protected void completed()
771 _callback.succeeded();
775 public void failed(final Throwable x)
778 failedCallback(_callback,x);
785 // Do a direct close of the output, as this may indicate to a client that the
786 // response is bad either with RST or by abnormal completion of chunked response.
787 getEndPoint().close();