Merge pull request #6657 from eclipse/jetty-10.0.x-6642-WebSocketConn…

…ectionHeaders

Issue #6642 - WebSocket handling of Connection: upgrade,close.

jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java

@@ -103,7 +103,7 @@ public void exchangeTerminated(HttpExchange exchange, Result result)
             closeReason = "failure";
         else if (receiver.isShutdown())
             closeReason = "server close";
-        else if (sender.isShutdown())
+        else if (sender.isShutdown() && response.getStatus() != HttpStatus.SWITCHING_PROTOCOLS_101)
             closeReason = "client close";
 
         if (closeReason == null)

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

@@ -344,7 +344,6 @@ private Result completing(ByteBuffer chunk, ByteBuffer content)
             _endOfContent = EndOfContent.UNKNOWN_CONTENT;
             return Result.FLUSH;
         }
-
         _state = State.END;
         return Boolean.TRUE.equals(_persistent) ? Result.DONE : Result.SHUTDOWN_OUT;
     }

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

@@ -893,8 +893,10 @@ private void releaseChunk()
         @Override
         protected void onCompleteSuccess()
         {
+            boolean upgrading = _channel.getRequest().getAttribute(UPGRADE_CONNECTION_ATTRIBUTE) != null;
             release().succeeded();
-            if (_shutdownOut)
+            // If successfully upgraded it is responsibility of the next protocol to close the connection.
+            if (_shutdownOut && !upgrading)
                 getEndPoint().shutdownOutput();
         }
 

jetty-websocket/websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/ConnectionHeaderTest.java

@@ -0,0 +1,101 @@
+//
+// ========================================================================
+// Copyright (c) 1995-2021 Mort Bay Consulting Pty Ltd and others.
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License v. 2.0 which is available at
+// https://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+// ========================================================================
+//
+
+package org.eclipse.jetty.websocket.tests;
+
+import java.net.URI;
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.jetty.client.HttpRequest;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.client.JettyUpgradeListener;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.eclipse.jetty.websocket.server.config.JettyWebSocketServletContainerInitializer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+public class ConnectionHeaderTest
+{
+    private static WebSocketClient client;
+    private static Server server;
+    private static ServerConnector connector;
+
+    @BeforeAll
+    public static void startContainers() throws Exception
+    {
+        server = new Server();
+        connector = new ServerConnector(server);
+        server.addConnector(connector);
+
+        ServletContextHandler contextHandler = new ServletContextHandler();
+        contextHandler.setContextPath("/");
+        JettyWebSocketServletContainerInitializer.configure(contextHandler, (servletContext, container) ->
+            container.addMapping("/echo", EchoSocket.class));
+
+        server.setHandler(contextHandler);
+        server.start();
+
+        client = new WebSocketClient();
+        client.start();
+    }
+
+    @AfterAll
+    public static void stopContainers() throws Exception
+    {
+        client.stop();
+        server.stop();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"Upgrade", "keep-alive, Upgrade", "close, Upgrade"})
+    public void testConnectionKeepAlive(String connectionHeaderValue) throws Exception
+    {
+        URI uri = URI.create("ws://localhost:" + connector.getLocalPort() + "/echo");
+        JettyUpgradeListener upgradeListener = new JettyUpgradeListener()
+        {
+            @Override
+            public void onHandshakeRequest(HttpRequest request)
+            {
+                HttpFields fields = request.getHeaders();
+                if (!(fields instanceof HttpFields.Mutable))
+                    throw new IllegalStateException(fields.getClass().getName());
+
+                // Replace the default connection header value with a custom one.
+                HttpFields.Mutable headers = (HttpFields.Mutable)fields;
+                headers.put(HttpHeader.CONNECTION, connectionHeaderValue);
+            }
+        };
+
+        EventSocket clientEndpoint = new EventSocket();
+        try (Session session = client.connect(clientEndpoint, uri, null, upgradeListener).get(5, TimeUnit.SECONDS))
+        {
+            // Generate text frame
+            String msg = "this is an echo ... cho ... ho ... o";
+            session.getRemote().sendString(msg);
+
+            // Read frame (hopefully text frame)
+            String response = clientEndpoint.textMessages.poll(5, TimeUnit.SECONDS);
+            assertThat("Text Frame.status code", response, is(msg));
+        }
+    }
+}