Implementation of HttpServletRequest.upgrade

jetty-http/src/main/java/org/eclipse/jetty/http/HttpGenerator.java

@@ -465,6 +465,12 @@ else if (status == HttpStatus.NO_CONTENT_204 || status == HttpStatus.NOT_MODIFIE
         }
     }
 
+    public void servletUpgrade()
+    {
+        _noContentResponse = false;
+        _state = State.COMMITTED;
+    }
+
     private void prepareChunk(ByteBuffer chunk, int remaining)
     {
         // if we need CRLF add this to header

jetty-http/src/main/java/org/eclipse/jetty/http/HttpParser.java

@@ -1685,8 +1685,8 @@ protected boolean parseContent(ByteBuffer buffer)
                     {
                         _contentChunk = buffer.asReadOnlyBuffer();
 
-                        // limit content by expected size
-                        if (remaining > content)
+                        // limit content by expected size if _contentLength is >= 0 (i.e.: not infinite)
+                        if (_contentLength > -1 && remaining > content)
                         {
                             // We can cast remaining to an int as we know that it is smaller than
                             // or equal to length which is already an int.
@@ -1888,6 +1888,13 @@ public void reset()
         _headerComplete = false;
     }
 
+    public void servletUpgrade()
+    {
+        setState(State.CONTENT);
+        _endOfContent = EndOfContent.UNKNOWN_CONTENT;
+        _contentLength = -1;
+    }
+
     protected void setState(State state)
     {
         if (debugEnabled)

jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java

@@ -929,7 +929,7 @@ protected boolean sendResponse(MetaData.Response response, ByteBuffer content, b
             commit(response);
             _combinedListener.onResponseBegin(_request);
             _request.onResponseCommit();
-            
+
             // wrap callback to process 100 responses
             final int status = response.getStatus();
             final Callback committed = (status < HttpStatus.OK_200 && status >= HttpStatus.CONTINUE_100)

jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelOverHttp.java

@@ -66,6 +66,7 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
     // events like timeout: we get notified and either schedule onError or release the
     // blocking semaphore.
     private HttpInput.Content _content;
+    private boolean _servletUpgrade;
 
     public HttpChannelOverHttp(HttpConnection httpConnection, Connector connector, HttpConfiguration config, EndPoint endPoint, HttpTransport transport)
     {
@@ -262,10 +263,19 @@ private void markEarlyEOF()
     {
         if (LOG.isDebugEnabled())
             LOG.debug("received early EOF, content = {}", _content);
-        EofException failure = new EofException("Early EOF");
-        if (_content != null)
-            _content.failed(failure);
-        _content = new HttpInput.ErrorContent(failure);
+        if (_servletUpgrade)
+        {
+            if (_content != null)
+                _content.succeeded();
+            _content = EOF;
+        }
+        else
+        {
+            EofException failure = new EofException("Early EOF");
+            if (_content != null)
+                _content.failed(failure);
+            _content = new HttpInput.ErrorContent(failure);
+        }
     }
 
     @Override
@@ -555,6 +565,16 @@ public void recycle()
         if (_content != null && !_content.isSpecial())
             throw new AssertionError("unconsumed content: " + _content);
         _content = null;
+        _servletUpgrade = false;
+    }
+
+    public void servletUpgrade()
+    {
+        if (_content != null && (!_content.isSpecial() || !_content.isEof()))
+            throw new IllegalStateException("Cannot perform servlet upgrade with unconsumed content");
+        _content = null;
+        _servletUpgrade = true;
+        _httpConnection.getParser().servletUpgrade();
     }
 
     @Override

jetty-server/src/main/java/org/eclipse/jetty/server/Request.java

@@ -47,6 +47,7 @@
 import javax.servlet.ServletContext;
 import javax.servlet.ServletException;
 import javax.servlet.ServletInputStream;
+import javax.servlet.ServletOutputStream;
 import javax.servlet.ServletRequest;
 import javax.servlet.ServletRequestAttributeEvent;
 import javax.servlet.ServletRequestAttributeListener;
@@ -61,6 +62,7 @@
 import javax.servlet.http.HttpUpgradeHandler;
 import javax.servlet.http.Part;
 import javax.servlet.http.PushBuilder;
+import javax.servlet.http.WebConnection;
 
 import org.eclipse.jetty.http.BadMessageException;
 import org.eclipse.jetty.http.ComplianceViolation;
@@ -78,6 +80,7 @@
 import org.eclipse.jetty.http.HttpVersion;
 import org.eclipse.jetty.http.MetaData;
 import org.eclipse.jetty.http.MimeTypes;
+import org.eclipse.jetty.io.Connection;
 import org.eclipse.jetty.io.RuntimeIOException;
 import org.eclipse.jetty.server.handler.ContextHandler;
 import org.eclipse.jetty.server.handler.ContextHandler.Context;
@@ -2140,6 +2143,11 @@ public AsyncContext startAsync() throws IllegalStateException
     {
         if (_asyncNotSupportedSource != null)
             throw new IllegalStateException("!asyncSupported: " + _asyncNotSupportedSource);
+        return forceStartAsync();
+    }
+
+    private AsyncContextState forceStartAsync()
+    {
         HttpChannelState state = getHttpChannelState();
         if (_async == null)
             _async = new AsyncContextState(state);
@@ -2372,7 +2380,95 @@ else if (oldQueryParams == null || oldQueryParams.size() == 0)
     @Override
     public <T extends HttpUpgradeHandler> T upgrade(Class<T> handlerClass) throws IOException, ServletException
     {
-        throw new ServletException("HttpServletRequest.upgrade() not supported in Jetty");
+        Response response = _channel.getResponse();
+        if (response.getStatus() != HttpStatus.SWITCHING_PROTOCOLS_101)
+            throw new IllegalStateException("Response status should be 101");
+        if (response.getHeader("Upgrade") == null)
+            throw new IllegalStateException("Missing Upgrade header");
+        if (!"Upgrade".equalsIgnoreCase(response.getHeader("Connection")))
+            throw new IllegalStateException("Invalid Connection header");
+        if (response.isCommitted())
+            throw new IllegalStateException("Cannot upgrade committed response");
+        if (_metaData == null || _metaData.getHttpVersion() != HttpVersion.HTTP_1_1)
+            throw new IllegalStateException("Only requests over HTTP/1.1 can be upgraded");
+
+        ServletOutputStream outputStream = response.getOutputStream();
+        ServletInputStream inputStream = getInputStream();
+        HttpChannelOverHttp httpChannel11 = (HttpChannelOverHttp)_channel;
+        HttpConnection httpConnection = (HttpConnection)_channel.getConnection();
+
+        T upgradeHandler;
+        try
+        {
+            upgradeHandler = handlerClass.getDeclaredConstructor().newInstance();
+        }
+        catch (Exception e)
+        {
+            throw new ServletException("Unable to instantiate handler class", e);
+        }
+
+        httpChannel11.servletUpgrade(); // tell the HTTP 1.1 channel that it is now handling an upgraded servlet
+        AsyncContext asyncContext = forceStartAsync(); // force the servlet in async mode
+
+        outputStream.flush(); // commit the 101 response
+        httpConnection.getGenerator().servletUpgrade(); // tell the generator it can send data as-is
+        httpConnection.addEventListener(new Connection.Listener()
+        {
+            @Override
+            public void onClosed(Connection connection)
+            {
+                try
+                {
+                    asyncContext.complete();
+                }
+                catch (Exception e)
+                {
+                    LOG.warn("error during upgrade AsyncContext complete", e);
+                }
+                try
+                {
+                    upgradeHandler.destroy();
+                }
+                catch (Exception e)
+                {
+                    LOG.warn("error during upgrade HttpUpgradeHandler destroy", e);
+                }
+            }
+
+            @Override
+            public void onOpened(Connection connection)
+            {
+            }
+        });
+
+        upgradeHandler.init(new WebConnection()
+        {
+            @Override
+            public void close() throws Exception
+            {
+                try
+                {
+                    inputStream.close();
+                }
+                finally
+                {
+                    outputStream.close();
+                }
+            }
+
+            @Override
+            public ServletInputStream getInputStream()
+            {
+                return inputStream;
+            }
+
+            @Override
+            public ServletOutputStream getOutputStream()
+            {
+                return outputStream;
+            }
+        });
+        return upgradeHandler;
     }
 
     /**