Issue #2282 - More work on websocket server side tests

jetty-websocket/javax-websocket-client-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/DecoderReaderManySmallTest.java

@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.io.Reader;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -40,13 +41,19 @@ import javax.websocket.WebSocketContainer;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
+import org.eclipse.jetty.websocket.api.StatusCode;
+import org.eclipse.jetty.websocket.common.CloseInfo;
+import org.eclipse.jetty.websocket.common.OpCode;
+import org.eclipse.jetty.websocket.common.WebSocketFrame;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -110,60 +117,9 @@ public class DecoderReaderManySmallTest
         }
     }
 
-    private static class EventIdServer implements Runnable
-    {
-        private BlockheadServer server;
-        private IBlockheadServerConnection sconnection;
-        private CountDownLatch connectLatch = new CountDownLatch(1);
-
-        public EventIdServer(BlockheadServer server)
-        {
-            this.server = server;
-        }
-
-        @Override
-        public void run()
-        {
-            try
-            {
-                sconnection = server.accept();
-                sconnection.setSoTimeout(60000);
-                sconnection.upgrade();
-            }
-            catch (Exception e)
-            {
-                LOG.warn(e);
-            }
-            finally
-            {
-                connectLatch.countDown();
-            }
-        }
-
-        public void writeSequentialIds(int from, int to) throws IOException
-        {
-            for (int id = from; id < to; id++)
-            {
-                TextFrame frame = new TextFrame();
-                frame.setPayload(Integer.toString(id));
-                sconnection.write(frame);
-            }
-        }
-
-        public void close() throws IOException
-        {
-            sconnection.close();
-        }
-
-        public void awaitConnect() throws InterruptedException
-        {
-            connectLatch.await(1,TimeUnit.SECONDS);
-        }
-    }
-
     private static final Logger LOG = Log.getLogger(DecoderReaderManySmallTest.class);
 
-    private BlockheadServer server;
+    private static BlockheadServer server;
     private WebSocketContainer client;
 
     @Before
@@ -178,15 +134,15 @@ public class DecoderReaderManySmallTest
         ((LifeCycle)client).stop();
     }
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -194,15 +150,38 @@ public class DecoderReaderManySmallTest
     @Test
     public void testManyIds() throws Exception
     {
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         EventIdSocket ids = new EventIdSocket();
-        EventIdServer idserver = new EventIdServer(server);
-        new Thread(idserver).start();
         client.connectToServer(ids,server.getWsUri());
-        idserver.awaitConnect();
+
-        int from = 1000;
+        final int from = 1000;
-        int to = 2000;
+        final int to = 2000;
-        idserver.writeSequentialIds(from,to);
+
-        idserver.close();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
+        {
+            // Setup echo of frames on server side
+            serverConn.setIncomingFrameConsumer((frame) ->
+            {
+                WebSocketFrame wsFrame = (WebSocketFrame) frame;
+                if (wsFrame.getOpCode() == OpCode.TEXT)
+                {
+                    String msg = wsFrame.getPayloadAsUTF8();
+                    if (msg == "generate")
+                    {
+                        for (int id = from; id < to; id++)
+                        {
+                            TextFrame event = new TextFrame();
+                            event.setPayload(Integer.toString(id));
+                            serverConn.write(event);
+                        }
+                        serverConn.write(new CloseInfo(StatusCode.NORMAL).asFrame());
+                    }
+                }
+            });
+
             int count = from - to;
             ids.awaitClose();
             // collect seen ids
@@ -222,3 +201,4 @@ public class DecoderReaderManySmallTest
             }
         }
     }
+}

jetty-websocket/javax-websocket-client-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/DecoderReaderTest.java

@@ -27,6 +27,8 @@ import java.io.IOException;
 import java.io.Reader;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -49,13 +51,14 @@ import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.common.WebSocketFrame;
 import org.eclipse.jetty.websocket.common.frames.ContinuationFrame;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Ignore;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class DecoderReaderTest
@@ -125,6 +128,7 @@ public class DecoderReaderTest
     @ClientEndpoint(decoders = { QuotesDecoder.class })
     public static class QuotesSocket
     {
+        private static final Logger LOG = Log.getLogger(QuotesSocket.class);
         public LinkedBlockingQueue<Quotes> messageQueue = new LinkedBlockingQueue<>();
         private CountDownLatch closeLatch = new CountDownLatch(1);
 
@@ -137,12 +141,15 @@ public class DecoderReaderTest
         @OnMessage
         public synchronized void onMessage(Quotes msg)
         {
-            Integer h=hashCode();
             messageQueue.offer(msg);
-            System.out.printf("%x: Quotes from: %s%n",h,msg.author);
+            if(LOG.isDebugEnabled())
+            {
+                String hashcode = Integer.toHexString(Objects.hashCode(this));
+                LOG.debug("{}: Quotes from: {}", hashcode, msg.author);
                 for (String quote : msg.quotes)
                 {
-                System.out.printf("%x: - %s%n",h,quote);
+                    LOG.debug("{}: - {}", hashcode, quote);
+                }
             }
         }
 
@@ -152,37 +159,78 @@ public class DecoderReaderTest
         }
     }
 
-    private static class QuoteServer implements Runnable
+    private static BlockheadServer server;
-    {
+    private WebSocketContainer client;
-        private BlockheadServer server;
-        private IBlockheadServerConnection sconnection;
-        private CountDownLatch connectLatch = new CountDownLatch(1);
 
-        public QuoteServer(BlockheadServer server)
+    @Before
+    public void initClient()
     {
-            this.server = server;
+        client = ContainerProvider.getWebSocketContainer();
     }
     
-        @Override
+    @After
-        public void run()
+    public void stopClient() throws Exception
     {
-            try
+        ((LifeCycle)client).stop();
-            {
-                sconnection = server.accept();
-                sconnection.setSoTimeout(60000);
-                sconnection.upgrade();
     }
-            catch (Exception e)
+    
+    @BeforeClass
+    public static void startServer() throws Exception
     {
-                LOG.warn(e);
+        server = new BlockheadServer();
+        server.start();
     }
-            finally
+
+    @AfterClass
+    public static void stopServer() throws Exception
     {
-                connectLatch.countDown();
+        server.stop();
+    }
+
+    @Test
+    public void testSingleQuotes() throws Exception
+    {
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
+        QuotesSocket quoter = new QuotesSocket();
+        client.connectToServer(quoter,server.getWsUri());
+
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
+        {
+            writeQuotes(serverConn, "quotes-ben.txt");
+
+            Quotes quotes = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
+            Assert.assertThat("Quotes Author", quotes.author, is("Benjamin Franklin"));
+            Assert.assertThat("Quotes Count", quotes.quotes.size(), is(3));
         }
     }
 
-        public void writeQuotes(String filename) throws IOException
+    @Test
+    // @Ignore ("Quotes appear to be able to arrive in any order?")
+    public void testTwoQuotes() throws Exception
+    {
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
+        QuotesSocket quoter = new QuotesSocket();
+        client.connectToServer(quoter,server.getWsUri());
+
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
+        {
+            writeQuotes( serverConn,"quotes-ben.txt");
+            writeQuotes( serverConn,"quotes-twain.txt");
+            Quotes quotes = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
+            Assert.assertThat("Quotes Author", quotes.author, is("Benjamin Franklin"));
+            Assert.assertThat("Quotes Count", quotes.quotes.size(), is(3));
+            quotes = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
+            Assert.assertThat("Quotes Author", quotes.author, is("Mark Twain"));
+        }
+    }
+
+    private void writeQuotes(BlockheadConnection conn, String filename) throws IOException
     {
         // read file
         File qfile = MavenTestingUtils.getTestResourceFile(filename);
@@ -210,87 +258,7 @@ public class DecoderReaderTest
             }
             frame.setFin((i >= (lines.size() - 1)));
             frame.setPayload(BufferUtil.toBuffer(lines.get(i) + "\n"));
-                sconnection.write(frame);
+            conn.write(frame);
         }
     }
-
-        public void close() throws IOException
-        {
-            sconnection.close();
-        }
-
-        public void awaitConnect() throws InterruptedException
-        {
-            connectLatch.await(1,TimeUnit.SECONDS);
-        }
-    }
-
-    private static final Logger LOG = Log.getLogger(DecoderReaderTest.class);
-
-    private BlockheadServer server;
-    private WebSocketContainer client;
-
-    @Before
-    public void initClient()
-    {
-        client = ContainerProvider.getWebSocketContainer();
-    }
-    
-    @After
-    public void stopClient() throws Exception
-    {
-        ((LifeCycle)client).stop();
-    }
-    
-    @Before
-    public void startServer() throws Exception
-    {
-        server = new BlockheadServer();
-        server.start();
-    }
-
-    @After
-    public void stopServer() throws Exception
-    {
-        server.stop();
-    }
-
-    // TODO analyse and fix
-    @Ignore
-    @Test
-    public void testSingleQuotes() throws Exception
-    {
-        QuotesSocket quoter = new QuotesSocket();
-        QuoteServer qserver = new QuoteServer(server);
-        new Thread(qserver).start();
-        client.connectToServer(quoter,server.getWsUri());
-        qserver.awaitConnect();
-        qserver.writeQuotes("quotes-ben.txt");
-        qserver.close();
-        quoter.awaitClose();
-        Quotes quotes = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
-        Assert.assertThat("Quotes Author",quotes.author,is("Benjamin Franklin"));
-        Assert.assertThat("Quotes Count",quotes.quotes.size(),is(3));
-    }
-
-    // TODO analyse and fix
-    @Test
-    @Ignore ("Quotes appear to be able to arrive in any order?")
-    public void testTwoQuotes() throws Exception
-    {
-        QuotesSocket quoter = new QuotesSocket();
-        QuoteServer qserver = new QuoteServer(server);
-        new Thread(qserver).start();
-        client.connectToServer(quoter,server.getWsUri());
-        qserver.awaitConnect();
-        qserver.writeQuotes("quotes-ben.txt");
-        qserver.writeQuotes("quotes-twain.txt");
-        qserver.close();
-        quoter.awaitClose();
-        Quotes quotes = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
-        Assert.assertThat("Quotes Author",quotes.author,is("Benjamin Franklin"));
-        Assert.assertThat("Quotes Count",quotes.quotes.size(),is(3));
-        quotes = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
-        Assert.assertThat("Quotes Author",quotes.author,is("Mark Twain"));
-    }
 }

jetty-websocket/javax-websocket-client-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/EncoderTest.java

@@ -27,8 +27,8 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Consumer;
 
 import javax.websocket.ClientEndpointConfig;
 import javax.websocket.ContainerProvider;
@@ -44,65 +44,21 @@ import org.eclipse.jetty.toolchain.test.MavenTestingUtils;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
+import org.eclipse.jetty.websocket.api.extensions.Frame;
+import org.eclipse.jetty.websocket.common.OpCode;
+import org.eclipse.jetty.websocket.common.WebSocketFrame;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class EncoderTest
 {
-    private static class EchoServer
-    {
-        private BlockheadServer server;
-        private IBlockheadServerConnection sconnection;
-        private CountDownLatch connectLatch = new CountDownLatch(1);
-
-        public EchoServer(BlockheadServer server)
-        {
-            this.server = server;
-        }
-
-        public void start()
-        {
-            CompletableFuture.runAsync(() -> {
-                try
-                {
-                    sconnection = server.accept();
-                    sconnection.setSoTimeout(10000);
-                    sconnection.upgrade();
-                    sconnection.enableIncomingEcho(true);
-                    sconnection.startReadThread();
-                }
-                catch (Exception e)
-                {
-                    LOG.warn(e);
-                }
-                finally
-                {
-                    connectLatch.countDown();
-                }
-            });
-        }
-
-        public void stop()
-        {
-            if (this.sconnection != null)
-            {
-                try
-                {
-                    this.sconnection.close();
-                }
-                catch (IOException ignore)
-                {
-                    /* ignore */
-                }
-            }
-        }
-    }
-
     public static class Quotes
     {
         private String author;
@@ -184,10 +140,22 @@ public class EncoderTest
 
     private static final Logger LOG = Log.getLogger(EncoderTest.class);
 
-    private BlockheadServer server;
+    private static BlockheadServer server;
-
     private WebSocketContainer client;
 
+    @BeforeClass
+    public static void startServer() throws Exception
+    {
+        server = new BlockheadServer();
+        server.start();
+    }
+
+    @AfterClass
+    public static void stopServer() throws Exception
+    {
+        server.stop();
+    }
+
     private void assertReceivedQuotes(String result, Quotes quotes)
     {
         Assert.assertThat("Quote Author",result,containsString("Author: " + quotes.getAuthor()));
@@ -236,26 +204,12 @@ public class EncoderTest
         ((LifeCycle)client).stop();
     }
 
-    @Before
-    public void startServer() throws Exception
-    {
-        server = new BlockheadServer();
-        server.start();
-    }
-
-    @After
-    public void stopServer() throws Exception
-    {
-        server.stop();
-    }
-
     @Test(timeout = 10000)
     public void testSingleQuotes() throws Exception
     {
-        EchoServer eserver = new EchoServer(server);
+        // Hook into server connection creation
-        try
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
-        {
+        server.addConnectFuture(serverConnFut);
-            eserver.start();
 
         QuotesSocket quoter = new QuotesSocket();
 
@@ -266,25 +220,25 @@ public class EncoderTest
         ClientEndpointConfig cec = builder.build();
         client.connectToServer(quoter,cec,server.getWsUri());
 
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
+        {
+            // Setup echo of frames on server side
+            serverConn.setIncomingFrameConsumer(new DataFrameEcho(serverConn));
+
             Quotes ben = getQuotes("quotes-ben.txt");
             quoter.write(ben);
 
             String result = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
             assertReceivedQuotes(result,ben);
         }
-        finally
-        {
-            eserver.stop();
-        }
     }
 
     @Test(timeout = 10000)
     public void testTwoQuotes() throws Exception
     {
-        EchoServer eserver = new EchoServer(server);
+        // Hook into server connection creation
-        try
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
-        {
+        server.addConnectFuture(serverConnFut);
-            eserver.start();
 
         QuotesSocket quoter = new QuotesSocket();
         ClientEndpointConfig.Builder builder = ClientEndpointConfig.Builder.create();
@@ -294,6 +248,11 @@ public class EncoderTest
         ClientEndpointConfig cec = builder.build();
         client.connectToServer(quoter,cec,server.getWsUri());
 
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
+        {
+            // Setup echo of frames on server side
+            serverConn.setIncomingFrameConsumer(new DataFrameEcho(serverConn));
+
             Quotes ben = getQuotes("quotes-ben.txt");
             Quotes twain = getQuotes("quotes-twain.txt");
             quoter.write(ben);
@@ -304,9 +263,26 @@ public class EncoderTest
             result = quoter.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
             assertReceivedQuotes(result,twain);
         }
-        finally
+    }
+
+    private static class DataFrameEcho implements Consumer<Frame>
     {
-            eserver.stop();
+        private final BlockheadConnection connection;
+
+        public DataFrameEcho(BlockheadConnection connection)
+        {
+            this.connection = connection;
+        }
+
+        @Override
+        public void accept(Frame frame)
+        {
+            if (OpCode.isDataFrame(frame.getOpCode()))
+            {
+                WebSocketFrame copy = WebSocketFrame.copy(frame);
+                copy.setMask(null); // remove client masking
+                connection.write(copy);
+            }
         }
     }
 }

jetty-websocket/websocket-client/pom.xml

@@ -40,6 +40,12 @@
             <artifactId>websocket-common</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.jetty.websocket</groupId>
             <artifactId>websocket-common</artifactId>

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/BadNetworkTest.java

@@ -19,6 +19,7 @@
 package org.eclipse.jetty.websocket.client;
 
 import java.net.URI;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -26,10 +27,13 @@ import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.io.MappedByteBufferPool;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
+import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 /**
@@ -39,7 +43,7 @@ public class BadNetworkTest
 {
     public ByteBufferPool bufferPool = new MappedByteBufferPool();
 
-    private BlockheadServer server;
+    private static BlockheadServer server;
     private WebSocketClient client;
 
     @Before
@@ -50,8 +54,8 @@ public class BadNetworkTest
         client.start();
     }
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
@@ -63,8 +67,8 @@ public class BadNetworkTest
         client.stop();
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -77,12 +81,9 @@ public class BadNetworkTest
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection ssocket = server.accept();
-        ssocket.upgrade();
-
         // Validate that we are connected
         future.get(30,TimeUnit.SECONDS);
-        wsocket.waitForConnected(30,TimeUnit.SECONDS);
+        wsocket.waitForConnected();
 
         // Have client disconnect abruptly
         Session session = wsocket.getSession();
@@ -102,18 +103,20 @@ public class BadNetworkTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection ssocket = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        ssocket.upgrade();
+        {
-
             // Validate that we are connected
             future.get(30, TimeUnit.SECONDS);
-        wsocket.waitForConnected(30,TimeUnit.SECONDS);
+            wsocket.waitForConnected();
 
             // Have server disconnect abruptly
-        ssocket.disconnect();
+            serverConn.abort();
 
             // Wait for close (as response to idle timeout)
             wsocket.waitForClose(10, TimeUnit.SECONDS);
@@ -124,3 +127,4 @@ public class BadNetworkTest
             wsocket.assertCloseCode(StatusCode.NO_CLOSE);
         }
     }
+}

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ClientCloseTest.java

@@ -34,7 +34,10 @@ import java.nio.ByteBuffer;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -67,13 +70,15 @@ import org.eclipse.jetty.websocket.common.WebSocketFrame;
 import org.eclipse.jetty.websocket.common.WebSocketSession;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
 import org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.RawFrameBuilder;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.hamcrest.Matcher;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -176,10 +181,10 @@ public class ClientCloseTest
         }
     }
 
-    private BlockheadServer server;
+    private static BlockheadServer server;
     private WebSocketClient client;
 
-    private void confirmConnection(CloseTrackingSocket clientSocket, Future<Session> clientFuture, IBlockheadServerConnection serverConns) throws Exception
+    private void confirmConnection(CloseTrackingSocket clientSocket, Future<Session> clientFuture, BlockheadConnection serverConns) throws Exception
     {
         // Wait for client connect on via future
         clientFuture.get(30,TimeUnit.SECONDS);
@@ -193,10 +198,8 @@ public class ClientCloseTest
             final String echoMsg = "echo-test";
             Future<Void> testFut = clientSocket.getRemote().sendStringByFuture(echoMsg);
 
-            serverConns.startReadThread();
-
             // Wait for send future
-            testFut.get(30,TimeUnit.SECONDS);
+            testFut.get(Timeouts.SEND, Timeouts.SEND_UNIT);
 
             // Read Frame on server side
             LinkedBlockingQueue<WebSocketFrame> serverCapture = serverConns.getFrameQueue();
@@ -220,7 +223,7 @@ public class ClientCloseTest
         }
     }
 
-    private void confirmServerReceivedCloseFrame(IBlockheadServerConnection serverConn, int expectedCloseCode, Matcher<String> closeReasonMatcher) throws InterruptedException
+    private void confirmServerReceivedCloseFrame(BlockheadConnection serverConn, int expectedCloseCode, Matcher<String> closeReasonMatcher) throws InterruptedException
     {
         LinkedBlockingQueue<WebSocketFrame> serverCapture = serverConn.getFrameQueue();
         WebSocketFrame frame = serverCapture.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
@@ -282,8 +285,8 @@ public class ClientCloseTest
         client.start();
     }
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
@@ -295,8 +298,8 @@ public class ClientCloseTest
         client.stop();
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -308,15 +311,15 @@ public class ClientCloseTest
         final int timeout = 5000;
         client.setMaxIdleTimeout(timeout);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CloseTrackingSocket clientSocket = new CloseTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
-        serverConn.upgrade();
-
-        try
         {
             // client confirms connection via echo
             confirmConnection(clientSocket, clientConnectFuture, serverConn);
@@ -348,10 +351,6 @@ public class ClientCloseTest
             // client close event on ws-endpoint
             clientSocket.assertReceivedCloseEvent(timeout, is(StatusCode.NORMAL), containsString("From Server"));
         }
-        finally
-        {
-            serverConn.disconnect();
-        }
     }
 
     @Ignore("Need sbordet's help here")
@@ -362,15 +361,15 @@ public class ClientCloseTest
         final int timeout = 1000;
         client.setMaxIdleTimeout(timeout);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CloseTrackingSocket clientSocket = new CloseTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
-        serverConn.upgrade();
-
-        try
         {
             // client confirms connection via echo
             confirmConnection(clientSocket, clientConnectFuture, serverConn);
@@ -402,10 +401,6 @@ public class ClientCloseTest
             assertThat("OnError Latch", clientSocket.errorLatch.await(2, TimeUnit.SECONDS), is(true));
             assertThat("OnError", clientSocket.error.get(), instanceOf(SocketTimeoutException.class));
         }
-        finally
-        {
-            serverConn.disconnect();
-        }
     }
 
     @Test
@@ -415,15 +410,15 @@ public class ClientCloseTest
         final int timeout = 1000;
         client.setMaxIdleTimeout(timeout);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CloseTrackingSocket clientSocket = new CloseTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
-        serverConn.upgrade();
-
-        try
         {
             // client confirms connection via echo
             confirmConnection(clientSocket, clientConnectFuture, serverConn);
@@ -440,9 +435,10 @@ public class ClientCloseTest
             bad.putShort((short) StatusCode.NORMAL);
             bad.put(msg);
             BufferUtil.flipToFlush(bad, 0);
-            try (StacklessLogging quiet = new StacklessLogging(Parser.class))
+
+            try (StacklessLogging ignore = new StacklessLogging(Parser.class))
             {
-                serverConn.write(bad);
+                serverConn.writeRaw(bad);
 
                 // client should have noticed the error
                 assertThat("OnError Latch", clientSocket.errorLatch.await(2, TimeUnit.SECONDS), is(true));
@@ -453,11 +449,6 @@ public class ClientCloseTest
                 confirmServerReceivedCloseFrame(serverConn, StatusCode.PROTOCOL, allOf(containsString("Invalid control frame"), containsString("length")));
             }
         }
-        finally
-        {
-            // server disconnects
-            serverConn.disconnect();
-        }
 
         // client triggers close event on client ws-endpoint
         clientSocket.assertReceivedCloseEvent(timeout,is(StatusCode.PROTOCOL),allOf(containsString("Invalid control frame"),containsString("length")));
@@ -470,15 +461,15 @@ public class ClientCloseTest
         final int timeout = 1000;
         client.setMaxIdleTimeout(timeout);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CloseTrackingSocket clientSocket = new CloseTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
-        serverConn.upgrade();
-
-        try
         {
             // client confirms connection via echo
             confirmConnection(clientSocket, clientConnectFuture, serverConn);
@@ -494,7 +485,7 @@ public class ClientCloseTest
             clientSocket.assertNoCloseEvent();
 
             // server shuts down connection (no frame reply)
-            serverConn.disconnect();
+            serverConn.abort();
 
             // client reads -1 (EOF)
             // client triggers close event on client ws-endpoint
@@ -504,10 +495,6 @@ public class ClientCloseTest
                             containsString("Disconnected")
                     ));
         }
-        finally
-        {
-            serverConn.disconnect();
-        }
     }
 
     @Test
@@ -517,15 +504,15 @@ public class ClientCloseTest
         final int timeout = 1000;
         client.setMaxIdleTimeout(timeout);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CloseTrackingSocket clientSocket = new CloseTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
-        serverConn.upgrade();
-
-        try
         {
             // client confirms connection via echo
             confirmConnection(clientSocket, clientConnectFuture, serverConn);
@@ -547,10 +534,6 @@ public class ClientCloseTest
             assertThat("OnError Latch", clientSocket.errorLatch.await(2, TimeUnit.SECONDS), is(true));
             assertThat("OnError", clientSocket.error.get(), instanceOf(TimeoutException.class));
         }
-        finally
-        {
-            serverConn.disconnect();
-        }
     }
 
     @Test(timeout = 5000L)
@@ -561,47 +544,58 @@ public class ClientCloseTest
         client.setMaxIdleTimeout(timeout);
 
         int clientCount = 3;
-        CloseTrackingSocket clientSockets[] = new CloseTrackingSocket[clientCount];
+        List<CloseTrackingSocket> clientSockets = new ArrayList<>();
-        IBlockheadServerConnection serverConns[] = new IBlockheadServerConnection[clientCount];
+        List<CompletableFuture<BlockheadConnection>> serverConnFuts = new ArrayList<>();
+        List<BlockheadConnection> serverConns = new ArrayList<>();
 
         try
         {
-            // Connect Multiple Clients
+            // Open Multiple Clients
             for (int i = 0; i < clientCount; i++)
             {
                 // Client Request Upgrade
-                clientSockets[i] = new CloseTrackingSocket();
+                CloseTrackingSocket clientSocket = new CloseTrackingSocket();
-                Future<Session> clientConnectFuture = client.connect(clientSockets[i], server.getWsUri());
+                clientSockets.add(clientSocket);
+                Future<Session> clientConnectFuture = client.connect(clientSocket, server.getWsUri());
 
                 // Server accepts connection
-                serverConns[i] = server.accept();
+                CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
-                serverConns[i].upgrade();
+                serverConnFuts.add(serverConnFut);
+                server.addConnectFuture(serverConnFut);
+                BlockheadConnection serverConn = serverConnFut.get();
+                serverConns.add(serverConn);
 
                 // client confirms connection via echo
-                confirmConnection(clientSockets[i], clientConnectFuture, serverConns[i]);
+                confirmConnection(clientSocket, clientConnectFuture, serverConn);
             }
 
-            // client lifecycle stop
+            // client lifecycle stop (the meat of this test)
             client.stop();
 
             // clients send close frames (code 1001, shutdown)
             for (int i = 0; i < clientCount; i++)
             {
                 // server receives close frame
-                confirmServerReceivedCloseFrame(serverConns[i], StatusCode.SHUTDOWN, containsString("Shutdown"));
+                confirmServerReceivedCloseFrame(serverConns.get(i), StatusCode.SHUTDOWN, containsString("Shutdown"));
             }
 
             // clients disconnect
             for (int i = 0; i < clientCount; i++)
             {
-                clientSockets[i].assertReceivedCloseEvent(timeout, is(StatusCode.SHUTDOWN), containsString("Shutdown"));
+                clientSockets.get(i).assertReceivedCloseEvent(timeout, is(StatusCode.SHUTDOWN), containsString("Shutdown"));
             }
         }
         finally
         {
-            for(IBlockheadServerConnection serverConn: serverConns)
+            for(BlockheadConnection serverConn: serverConns)
             {
-                serverConn.disconnect();
+                try
+                {
+                    serverConn.close();
+                }
+                catch (Exception ignore)
+                {
+                }
             }
         }
     }
@@ -613,15 +607,15 @@ public class ClientCloseTest
         final int timeout = 1000;
         client.setMaxIdleTimeout(timeout);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CloseTrackingSocket clientSocket = new CloseTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
-        serverConn.upgrade();
-
-        try
         {
             // client confirms connection via echo
             confirmConnection(clientSocket, clientConnectFuture, serverConn);
@@ -643,9 +637,5 @@ public class ClientCloseTest
             // assert - close reason message contains (write failure)
             clientSocket.assertReceivedCloseEvent(timeout, is(StatusCode.ABNORMAL), containsString("EOF"));
         }
-        finally
-        {
-            serverConn.disconnect();
-        }
     }
 }

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ClientConnectTest.java

@@ -27,27 +27,37 @@ import static org.junit.Assert.assertThat;
 
 import java.io.IOException;
 import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
 import java.net.SocketTimeoutException;
 import java.net.URI;
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import javax.servlet.http.HttpServletResponse;
+
 import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.io.MappedByteBufferPool;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.UpgradeException;
 import org.eclipse.jetty.websocket.common.AcceptHash;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
+import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.hamcrest.Matcher;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 /**
@@ -58,8 +68,7 @@ public class ClientConnectTest
 {
     public ByteBufferPool bufferPool = new MappedByteBufferPool();
 
-    private final int timeout = 500;
+    private static BlockheadServer server;
-    private BlockheadServer server;
     private WebSocketClient client;
 
     @SuppressWarnings("unchecked")
@@ -88,25 +97,32 @@ public class ClientConnectTest
     {
         client = new WebSocketClient();
         client.setBufferPool(bufferPool);
-        client.setConnectTimeout(timeout);
+        client.setConnectTimeout(Timeouts.CONNECT_UNIT.toMillis(Timeouts.CONNECT));
         client.start();
     }
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
     }
 
+    @Before
+    public void resetServerHandler()
+    {
+        // for each test, reset the server request handling to default
+        server.resetRequestHandling();
+    }
+
     @After
     public void stopClient() throws Exception
     {
         client.stop();
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -119,12 +135,9 @@ public class ClientConnectTest
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        connection.upgrade();
-
         Session sess = future.get(30,TimeUnit.SECONDS);
         
-        wsocket.waitForConnected(1, TimeUnit.SECONDS);
+        wsocket.waitForConnected();
         
         assertThat("Connect.UpgradeRequest", wsocket.connectUpgradeRequest, notNullValue());
         assertThat("Connect.UpgradeResponse", wsocket.connectUpgradeResponse, notNullValue());
@@ -139,6 +152,8 @@ public class ClientConnectTest
         URI wsUri = server.getWsUri();
         
         HttpClient httpClient = new HttpClient();
+        try
+        {
             httpClient.start();
 
             WebSocketUpgradeRequest req = new WebSocketUpgradeRequest(new WebSocketClient(), httpClient, wsUri, wsocket);
@@ -151,9 +166,11 @@ public class ClientConnectTest
                 assertThat("Connect.UpgradeRequest", wsocket.connectUpgradeRequest, notNullValue());
                 assertThat("Connect.UpgradeResponse", wsocket.connectUpgradeResponse, notNullValue());
             });
-        
+        }
-        IBlockheadServerConnection connection = server.accept();
+        finally
-        connection.upgrade();
+        {
+            httpClient.stop();
+        }
     }
     
     @Test
@@ -161,6 +178,10 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         ClientUpgradeRequest upgradeRequest = new ClientUpgradeRequest();
         // actual value for this test is irrelevant, its important that this
@@ -168,36 +189,35 @@ public class ClientConnectTest
         upgradeRequest.setHeader("Authorization", "Bogus SHA1");
         Future<Session> future = client.connect(wsocket,wsUri,upgradeRequest);
 
-        IBlockheadServerConnection connection = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        List<String> requestLines = connection.upgrade();
+        {
+            HttpFields upgradeRequestHeaders = serverConn.getUpgradeRequestHeaders();
 
             Session sess = future.get(30, TimeUnit.SECONDS);
             sess.close();
 
-        String authLine = requestLines.stream()
+            HttpField authHeader = upgradeRequestHeaders.getField(HttpHeader.AUTHORIZATION);
-                .filter((line) -> line.startsWith("Authorization:"))
+            assertThat("Server Request Authorization Header", authHeader, is(notNullValue()));
-                .findFirst().get();
+            assertThat("Server Request Authorization Value", authHeader.getValue(), is("Authorization: Bogus SHA1"));
-        
-        assertThat("Request Container Authorization", authLine, is("Authorization: Bogus SHA1"));
             assertThat("Connect.UpgradeRequest", wsocket.connectUpgradeRequest, notNullValue());
             assertThat("Connect.UpgradeResponse", wsocket.connectUpgradeResponse, notNullValue());
         }
+    }
     
     @Test
     public void testBadHandshake() throws Exception
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Force 404 response, no upgrade for this test
+        server.setRequestHandling((req, resp) -> {
+            resp.setStatus(HttpServletResponse.SC_NOT_FOUND);
+            return true;
+        });
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        connection.readRequest();
-        // no upgrade, just fail with a 404 error
-        connection.respond("HTTP/1.1 404 NOT FOUND\r\n" +
-                "Content-Length: 0\r\n" +
-                "\r\n");
-
         // The attempt to get upgrade response future should throw error
         try
         {
@@ -219,16 +239,15 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Force 200 response, no response body content, no upgrade for this test
+        server.setRequestHandling((req, resp) -> {
+            resp.setStatus(HttpServletResponse.SC_OK);
+            return true;
+        });
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        connection.readRequest();
-        // Send OK to GET but not upgrade
-        connection.respond("HTTP/1.1 200 OK\r\n" +
-                "Content-Length: 0\r\n" +
-                "\r\n");
-
         // The attempt to get upgrade response future should throw error
         try
         {
@@ -250,22 +269,17 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Force 200 response, no response body content, incomplete websocket response headers, no actual upgrade for this test
+        server.setRequestHandling((req, resp) -> {
+            String key = req.getHeader(HttpHeader.SEC_WEBSOCKET_KEY.toString());
+            resp.setStatus(HttpServletResponse.SC_OK);
+            resp.setHeader(HttpHeader.SEC_WEBSOCKET_ACCEPT.toString(), AcceptHash.hashKey(key));
+            return true;
+        });
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        List<String> requestLines = connection.readRequestLines();
-        String key = connection.parseWebSocketKey(requestLines);
-
-        // Send OK to GET but not upgrade
-        StringBuilder resp = new StringBuilder();
-        resp.append("HTTP/1.1 200 OK\r\n"); // intentionally 200 (not 101)
-        // Include a value accept key
-        resp.append("Sec-WebSocket-Accept: ").append(AcceptHash.hashKey(key)).append("\r\n");
-        resp.append("Content-Length: 0\r\n");
-        resp.append("\r\n");
-        connection.respond(resp.toString());
-
         // The attempt to get upgrade response future should throw error
         try
         {
@@ -287,21 +301,18 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Force 101 response, with invalid Connection header, invalid handshake
+        server.setRequestHandling((req, resp) -> {
+            String key = req.getHeader(HttpHeader.SEC_WEBSOCKET_KEY.toString());
+            resp.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
+            resp.setHeader(HttpHeader.CONNECTION.toString(), "close");
+            resp.setHeader(HttpHeader.SEC_WEBSOCKET_ACCEPT.toString(), AcceptHash.hashKey(key));
+            return true;
+        });
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        List<String> requestLines = connection.readRequestLines();
-        String key = connection.parseWebSocketKey(requestLines);
-
-        // Send Switching Protocols 101, but invalid 'Connection' header
-        StringBuilder resp = new StringBuilder();
-        resp.append("HTTP/1.1 101 Switching Protocols\r\n");
-        resp.append("Sec-WebSocket-Accept: ").append(AcceptHash.hashKey(key)).append("\r\n");
-        resp.append("Connection: close\r\n");
-        resp.append("\r\n");
-        connection.respond(resp.toString());
-
         // The attempt to get upgrade response future should throw error
         try
         {
@@ -323,21 +334,18 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Force 101 response, with no Connection header, invalid handshake
+        server.setRequestHandling((req, resp) -> {
+            String key = req.getHeader(HttpHeader.SEC_WEBSOCKET_KEY.toString());
+            resp.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
+            // Intentionally leave out Connection header
+            resp.setHeader(HttpHeader.SEC_WEBSOCKET_ACCEPT.toString(), AcceptHash.hashKey(key));
+            return true;
+        });
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        List<String> requestLines = connection.readRequestLines();
-        String key = connection.parseWebSocketKey(requestLines);
-
-        // Send Switching Protocols 101, but no 'Connection' header
-        StringBuilder resp = new StringBuilder();
-        resp.append("HTTP/1.1 101 Switching Protocols\r\n");
-        resp.append("Sec-WebSocket-Accept: ").append(AcceptHash.hashKey(key)).append("\r\n");
-        // Intentionally leave out Connection header
-        resp.append("\r\n");
-        connection.respond(resp.toString());
-
         // The attempt to get upgrade response future should throw error
         try
         {
@@ -359,14 +367,16 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
+        // Force 101 response, with invalid response accept header
+        server.setRequestHandling((req, resp) -> {
+            resp.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
+            resp.setHeader(HttpHeader.SEC_WEBSOCKET_ACCEPT.toString(), "rubbish");
+            return true;
+        });
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection connection = server.accept();
-        connection.readRequest();
-        // Upgrade badly
-        connection.respond("HTTP/1.1 101 Upgrade\r\n" + "Sec-WebSocket-Accept: rubbish\r\n" + "\r\n");
-
         // The attempt to get upgrade response future should throw error
         try
         {
@@ -388,11 +398,17 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
-        URI wsUri = server.getWsUri();
+        try(ServerSocket serverSocket = new ServerSocket())
+        {
+            InetAddress addr = InetAddress.getByName("localhost");
+            InetSocketAddress endpoint = new InetSocketAddress(addr, 0);
+            serverSocket.bind(endpoint, 1);
+            int port = serverSocket.getLocalPort();
+            URI wsUri = URI.create(String.format("ws://%s:%d/", addr.getHostAddress(), port));
             Future<Session> future = client.connect(wsocket, wsUri);
 
             // Intentionally not accept incoming socket.
-        // server.accept();
+            // serverSocket.accept();
 
             try
             {
@@ -411,6 +427,7 @@ public class ClientConnectTest
                 wsocket.assertNotOpened();
             }
         }
+    }
 
     @Test
     public void testConnectionRefused() throws Exception
@@ -449,13 +466,17 @@ public class ClientConnectTest
     {
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
 
-        URI wsUri = server.getWsUri();
+        try(ServerSocket serverSocket = new ServerSocket())
+        {
+            InetAddress addr = InetAddress.getByName("localhost");
+            InetSocketAddress endpoint = new InetSocketAddress(addr, 0);
+            serverSocket.bind(endpoint, 1);
+            int port = serverSocket.getLocalPort();
+            URI wsUri = URI.create(String.format("ws://%s:%d/", addr.getHostAddress(), port));
             Future<Session> future = client.connect(wsocket, wsUri);
 
-        IBlockheadServerConnection ssocket = server.accept();
+            // Accept the connection, but do nothing on it (no response, no upgrade, etc)
-        Assert.assertNotNull(ssocket);
+            serverSocket.accept();
-        // Intentionally don't upgrade
-        // ssocket.upgrade();
 
             // The attempt to get upgrade response future should throw error
             try
@@ -470,3 +491,4 @@ public class ClientConnectTest
             }
         }
     }
+}

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/CookieTest.java

@@ -25,25 +25,30 @@ import static org.junit.Assert.assertTrue;
 import java.net.CookieManager;
 import java.net.HttpCookie;
 import java.util.Collections;
-import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
 import org.eclipse.jetty.websocket.api.WebSocketAdapter;
-import org.eclipse.jetty.websocket.api.util.QuoteUtil;
+import org.eclipse.jetty.websocket.common.CloseInfo;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class CookieTest
@@ -83,8 +88,8 @@ public class CookieTest
         }
     }
 
+    private static BlockheadServer server;
     private WebSocketClient client;
-    private BlockheadServer server;
 
     @Before
     public void startClient() throws Exception
@@ -93,8 +98,8 @@ public class CookieTest
         client.start();
     }
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
@@ -109,8 +114,8 @@ public class CookieTest
         }
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -132,19 +137,23 @@ public class CookieTest
         cookie.setMaxAge(100000);
         cookieMgr.getCookieStore().add(server.getWsUri(),cookie);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CookieTrackingSocket clientSocket = new CookieTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
+        {
-
             // client confirms upgrade and receipt of frame
             String serverCookies = confirmClientUpgradeAndCookies(clientSocket, clientConnectFuture, serverConn);
 
             assertThat("Cookies seen at server side", serverCookies, containsString("hello=world"));
             assertThat("Cookies seen at server side", serverCookies, containsString("foo=bar is the word"));
         }
+    }
     
     @Test
     public void testViaServletUpgradeRequest() throws Exception
@@ -157,32 +166,35 @@ public class CookieTest
         ClientUpgradeRequest request = new ClientUpgradeRequest();
         request.setCookies(Collections.singletonList(cookie));
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         // Client connects
         CookieTrackingSocket clientSocket = new CookieTrackingSocket();
         Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri(),request);
 
-        // Server accepts connect
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        IBlockheadServerConnection serverConn = server.accept();
+        {
-
             // client confirms upgrade and receipt of frame
             String serverCookies = confirmClientUpgradeAndCookies(clientSocket, clientConnectFuture, serverConn);
 
             Assert.assertThat("Cookies seen at server side", serverCookies, containsString("hello=world"));
         }
+    }
 
-    private String confirmClientUpgradeAndCookies(CookieTrackingSocket clientSocket, Future<Session> clientConnectFuture, IBlockheadServerConnection serverConn)
+    private String confirmClientUpgradeAndCookies(CookieTrackingSocket clientSocket, Future<Session> clientConnectFuture, BlockheadConnection serverConn)
             throws Exception
     {
-        // Server upgrades
+        // Server side upgrade information
-        List<String> upgradeRequestLines = serverConn.upgrade();
+        HttpFields upgradeRequestHeaders = serverConn.getUpgradeRequestHeaders();
-        List<String> upgradeRequestCookies = serverConn.regexFind(upgradeRequestLines,"^Cookie: (.*)$");
+        HttpField cookieField = upgradeRequestHeaders.getField(HttpHeader.COOKIE);
 
         // Server responds with cookies it knows about
         TextFrame serverCookieFrame = new TextFrame();
         serverCookieFrame.setFin(true);
-        serverCookieFrame.setPayload(QuoteUtil.join(upgradeRequestCookies,","));
+        serverCookieFrame.setPayload(cookieField.getValue());
         serverConn.write(serverCookieFrame);
-        serverConn.flush();
 
         // Confirm client connect on future
         clientConnectFuture.get(10,TimeUnit.SECONDS);
@@ -193,7 +205,7 @@ public class CookieTest
         LOG.debug("Cookies seen at server: {}",cookies);
         
         // Server closes connection
-        serverConn.close(StatusCode.NORMAL);
+        serverConn.write(new CloseInfo(StatusCode.NORMAL).asFrame());
     
         return cookies;
     }

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/JettyTrackingSocket.java

@@ -34,6 +34,7 @@ import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.UpgradeRequest;
 import org.eclipse.jetty.websocket.api.UpgradeResponse;
 import org.eclipse.jetty.websocket.api.WebSocketAdapter;
+import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.Assert;
 
 /**
@@ -159,9 +160,9 @@ public class JettyTrackingSocket extends WebSocketAdapter
         Assert.assertThat("Client Socket Closed",closeLatch.await(timeoutDuration,timeoutUnit),is(true));
     }
 
-    public void waitForConnected(int timeoutDuration, TimeUnit timeoutUnit) throws InterruptedException
+    public void waitForConnected() throws InterruptedException
     {
-        Assert.assertThat("Client Socket Connected",openLatch.await(timeoutDuration,timeoutUnit),is(true));
+        Assert.assertThat("Client Socket Connected",openLatch.await(Timeouts.CONNECT,Timeouts.CONNECT_UNIT),is(true));
     }
 
     public void waitForMessage(int timeoutDuration, TimeUnit timeoutUnit) throws InterruptedException

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ServerReadThread.java

@@ -1,133 +0,0 @@
-//
-//  ========================================================================
-//  Copyright (c) 1995-2018 Mort Bay Consulting Pty. Ltd.
-//  ------------------------------------------------------------------------
-//  All rights reserved. This program and the accompanying materials
-//  are made available under the terms of the Eclipse Public License v1.0
-//  and Apache License v2.0 which accompanies this distribution.
-//
-//      The Eclipse Public License is available at
-//      http://www.eclipse.org/legal/epl-v10.html
-//
-//      The Apache License v2.0 is available at
-//      http://www.opensource.org/licenses/apache2.0.php
-//
-//  You may elect to redistribute this code under either of these licenses.
-//  ========================================================================
-//
-
-package org.eclipse.jetty.websocket.client;
-
-import static org.hamcrest.Matchers.is;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.eclipse.jetty.io.ByteBufferPool;
-import org.eclipse.jetty.util.BufferUtil;
-import org.eclipse.jetty.util.log.Log;
-import org.eclipse.jetty.util.log.Logger;
-import org.eclipse.jetty.websocket.common.CloseInfo;
-import org.eclipse.jetty.websocket.common.OpCode;
-import org.eclipse.jetty.websocket.common.WebSocketFrame;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
-import org.eclipse.jetty.websocket.common.test.Timeouts;
-import org.junit.Assert;
-
-public class ServerReadThread extends Thread
-{
-    private static final int BUFFER_SIZE = 8192;
-    private static final Logger LOG = Log.getLogger(ServerReadThread.class);
-    private final IBlockheadServerConnection conn;
-    private boolean active = true;
-    private int slowness = -1; // disabled is default
-    private final AtomicInteger frameCount = new AtomicInteger();
-    private final CountDownLatch expectedMessageCount;
-
-    public ServerReadThread(IBlockheadServerConnection sconnection, int expectedMessages)
-    {
-        this.conn = sconnection;
-        this.expectedMessageCount = new CountDownLatch(expectedMessages);
-    }
-
-    public void cancel()
-    {
-        active = false;
-    }
-
-    public int getFrameCount()
-    {
-        return frameCount.get();
-    }
-
-    public int getSlowness()
-    {
-        return slowness;
-    }
-
-    @Override
-    public void run()
-    {
-        ByteBufferPool bufferPool = conn.getBufferPool();
-        ByteBuffer buf = bufferPool.acquire(BUFFER_SIZE,false);
-        BufferUtil.clearToFill(buf);
-
-        try
-        {
-            while (active)
-            {
-                BufferUtil.clearToFill(buf);
-                int len = conn.read(buf);
-
-                if (len > 0)
-                {
-                    LOG.debug("Read {} bytes",len);
-                    BufferUtil.flipToFlush(buf,0);
-                    conn.getParser().parse(buf);
-                }
-
-                LinkedBlockingQueue<WebSocketFrame> frames = conn.getFrameQueue();
-                WebSocketFrame frame;
-                while ((frame = frames.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT)) != null)
-                {
-                    frameCount.incrementAndGet();
-                    if (frame.getOpCode() == OpCode.CLOSE)
-                    {
-                        active = false;
-                        // automatically response to close frame
-                        CloseInfo close = new CloseInfo(frame);
-                        conn.close(close.getStatusCode());
-                    }
-
-                    expectedMessageCount.countDown();
-                }
-                if (slowness > 0)
-                {
-                    TimeUnit.MILLISECONDS.sleep(getSlowness());
-                }
-            }
-        }
-        catch (IOException | InterruptedException e)
-        {
-            LOG.warn(e);
-        }
-        finally
-        {
-            bufferPool.release(buf);
-        }
-    }
-
-    public void setSlowness(int slowness)
-    {
-        this.slowness = slowness;
-    }
-
-    public void waitForExpectedMessageCount(int timeoutDuration, TimeUnit timeoutUnit) throws InterruptedException
-    {
-        Assert.assertThat("Expected Message Count attained",expectedMessageCount.await(timeoutDuration,timeoutUnit),is(true));
-    }
-}

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ServerWriteThread.java

@@ -18,24 +18,23 @@
 
 package org.eclipse.jetty.websocket.client;
 
-import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 
 public class ServerWriteThread extends Thread
 {
     private static final Logger LOG = Log.getLogger(ServerWriteThread.class);
-    private final IBlockheadServerConnection conn;
+    private final BlockheadConnection conn;
     private int slowness = -1;
     private int messageCount = 100;
     private String message = "Hello";
 
-    public ServerWriteThread(IBlockheadServerConnection conn)
+    public ServerWriteThread(BlockheadConnection conn)
     {
         this.conn = conn;
     }
@@ -74,7 +73,7 @@ public class ServerWriteThread extends Thread
                 }
             }
         }
-        catch (InterruptedException | IOException e)
+        catch (InterruptedException e)
         {
             LOG.warn(e);
         }

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/SessionTest.java

@@ -26,35 +26,37 @@ import static org.junit.Assert.assertThat;
 import java.net.URI;
 import java.util.Collection;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import org.eclipse.jetty.websocket.api.BatchMode;
 import org.eclipse.jetty.websocket.api.RemoteEndpoint;
 import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.common.WebSocketFrame;
 import org.eclipse.jetty.websocket.common.WebSocketSession;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
-import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
-import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
 
 public class SessionTest
 {
-    private BlockheadServer server;
+    private static BlockheadServer server;
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -69,6 +71,10 @@ public class SessionTest
         {
             JettyTrackingSocket cliSock = new JettyTrackingSocket();
 
+            // Hook into server connection creation
+            CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+            server.addConnectFuture(serverConnFut);
+
             client.getPolicy().setIdleTimeout(10000);
 
             URI wsUri = server.getWsUri();
@@ -76,8 +82,14 @@ public class SessionTest
             request.setSubProtocols("echo");
             Future<Session> future = client.connect(cliSock,wsUri,request);
 
-            final IBlockheadServerConnection srvSock = server.accept();
+            try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-            srvSock.upgrade();
+            {
+                // Setup echo of frames on server side
+                serverConn.setIncomingFrameConsumer((frame)->{
+                    WebSocketFrame copy = WebSocketFrame.copy(frame);
+                    serverConn.write(copy);
+                });
+
                 Session sess = future.get(30000, TimeUnit.MILLISECONDS);
                 Assert.assertThat("Session", sess, notNullValue());
                 Assert.assertThat("Session.open", sess.isOpen(), is(true));
@@ -97,10 +109,6 @@ public class SessionTest
                     remote.flush();
                 }
 
-            try
-            {
-                srvSock.enableIncomingEcho(true);
-                srvSock.startReadThread();
                 // wait for response from server
                 cliSock.waitForMessage(30000, TimeUnit.MILLISECONDS);
 
@@ -111,11 +119,6 @@ public class SessionTest
                 assertThat("Message", received, containsString("Hello World!"));
 
                 cliSock.close();
-                srvSock.close();
-            }
-            finally
-            {
-                srvSock.disconnect();
             }
 
             cliSock.waitForClose(30000, TimeUnit.MILLISECONDS);

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/SlowClientTest.java

@@ -19,23 +19,32 @@
 package org.eclipse.jetty.websocket.client;
 
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
 
 import java.net.URI;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
+import org.eclipse.jetty.websocket.common.CloseInfo;
+import org.eclipse.jetty.websocket.common.OpCode;
+import org.eclipse.jetty.websocket.common.WebSocketFrame;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
+import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class SlowClientTest
 {
-    private BlockheadServer server;
+    private static BlockheadServer server;
     private WebSocketClient client;
 
     @Before
@@ -46,8 +55,8 @@ public class SlowClientTest
         client.start();
     }
 
-    @Before
+    @BeforeClass
-    public void startServer() throws Exception
+    public static void startServer() throws Exception
     {
         server = new BlockheadServer();
         server.start();
@@ -59,8 +68,8 @@ public class SlowClientTest
         client.stop();
     }
 
-    @After
+    @AfterClass
-    public void stopServer() throws Exception
+    public static void stopServer() throws Exception
     {
         server.stop();
     }
@@ -74,20 +83,17 @@ public class SlowClientTest
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(tsocket, wsUri);
 
-        IBlockheadServerConnection sconnection = server.accept();
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
-        sconnection.setSoTimeout(60000);
+        server.addConnectFuture(serverConnFut);
-        sconnection.upgrade();
 
         // Confirm connected
-        future.get(30,TimeUnit.SECONDS);
+        future.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT);
-        tsocket.waitForConnected(30,TimeUnit.SECONDS);
+        tsocket.waitForConnected();
 
         int messageCount = 10;
 
-        // Setup server read thread
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        ServerReadThread reader = new ServerReadThread(sconnection, messageCount);
+        {
-        reader.start();
-
             // Have client write slowly.
             ClientWriteThread writer = new ClientWriteThread(tsocket.getSession());
             writer.setMessageCount(messageCount);
@@ -96,17 +102,31 @@ public class SlowClientTest
             writer.start();
             writer.join();
 
-        reader.waitForExpectedMessageCount(1, TimeUnit.MINUTES);
+            LinkedBlockingQueue<WebSocketFrame> serverFrames = serverConn.getFrameQueue();
 
-        // Verify receive
+            for (int i = 0; i < messageCount; i++)
-        Assert.assertThat("Frame Receive Count", reader.getFrameCount(), is(messageCount));
+            {
+                WebSocketFrame serverFrame = serverFrames.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
+                String prefix = "Server frame[" + i + "]";
+                assertThat(prefix + ".opcode", serverFrame.getOpCode(), is(OpCode.TEXT));
+                assertThat(prefix + ".payload", serverFrame.getPayloadAsUTF8(), is("Hello/" + i + "/"));
+            }
 
             // Close
             tsocket.getSession().close(StatusCode.NORMAL, "Done");
 
+            // confirm close received on server
+            WebSocketFrame serverFrame = serverFrames.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
+            assertThat("close frame", serverFrame.getOpCode(), is(OpCode.CLOSE));
+            CloseInfo closeInfo = new CloseInfo(serverFrame);
+            assertThat("close info", closeInfo.getStatusCode(), is(StatusCode.NORMAL));
+            WebSocketFrame respClose = WebSocketFrame.copy(serverFrame);
+            respClose.setMask(null); // remove client mask (if present)
+            serverConn.write(respClose);
+
+            // Verify server response
             Assert.assertTrue("Client Socket Closed", tsocket.closeLatch.await(3, TimeUnit.MINUTES));
             tsocket.assertCloseCode(StatusCode.NORMAL);
-
+        }
-        reader.cancel(); // stop reading
     }
 }

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/SlowServerTest.java

@@ -19,16 +19,22 @@
 package org.eclipse.jetty.websocket.client;
 
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 
 import java.net.URI;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
 import org.eclipse.jetty.websocket.client.masks.ZeroMasker;
+import org.eclipse.jetty.websocket.common.OpCode;
+import org.eclipse.jetty.websocket.common.WebSocketFrame;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
+import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -73,24 +79,31 @@ public class SlowServerTest
         client.setMasker(new ZeroMasker());
         client.setMaxIdleTimeout(60000);
 
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(tsocket,wsUri);
 
-        IBlockheadServerConnection sconnection = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        sconnection.setSoTimeout(60000);
+        {
-        sconnection.upgrade();
+            // slow down reads
+            serverConn.setIncomingFrameConsumer((frame)-> {
+                try
+                {
+                    TimeUnit.MILLISECONDS.sleep(100);
+                }
+                catch (InterruptedException ignore)
+                {
+                }
+            });
 
             // Confirm connected
-        future.get(30,TimeUnit.SECONDS);
+            future.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT);
-        tsocket.waitForConnected(30,TimeUnit.SECONDS);
+            tsocket.waitForConnected();
 
             int messageCount = 10;
 
-        // Setup slow server read thread
-        ServerReadThread reader = new ServerReadThread(sconnection, messageCount);
-        reader.setSlowness(100); // slow it down
-        reader.start();
-
             // Have client write as quickly as it can.
             ClientWriteThread writer = new ClientWriteThread(tsocket.getSession());
             writer.setMessageCount(messageCount);
@@ -100,35 +113,41 @@ public class SlowServerTest
             writer.join();
 
             // Verify receive
-        reader.waitForExpectedMessageCount(10,TimeUnit.SECONDS);
+            LinkedBlockingQueue<WebSocketFrame> serverFrames = serverConn.getFrameQueue();
-        Assert.assertThat("Frame Receive Count",reader.getFrameCount(),is(messageCount));
+            for(int i=0; i< messageCount; i++)
+            {
+                WebSocketFrame serverFrame = serverFrames.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
+                String prefix = "Server Frame[" + i + "]";
+                Assert.assertThat(prefix, serverFrame, is(notNullValue()));
+                Assert.assertThat(prefix + ".opCode", serverFrame.getOpCode(), is(OpCode.TEXT));
+                Assert.assertThat(prefix + ".payload", serverFrame.getPayloadAsUTF8(), is("Hello"));
+            }
 
             // Close
             tsocket.getSession().close(StatusCode.NORMAL, "Done");
 
             Assert.assertTrue("Client Socket Closed", tsocket.closeLatch.await(10, TimeUnit.SECONDS));
             tsocket.assertCloseCode(StatusCode.NORMAL);
-
+        }
-        reader.cancel(); // stop reading
     }
 
     @Test
     public void testServerSlowToSend() throws Exception
     {
         JettyTrackingSocket clientSocket = new JettyTrackingSocket();
-        client.setMasker(new ZeroMasker());
         client.setMaxIdleTimeout(60000);
 
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         Future<Session> clientConnectFuture = client.connect(clientSocket,wsUri);
 
-        IBlockheadServerConnection serverConn = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        serverConn.setSoTimeout(60000);
+        {
-        serverConn.upgrade();
-
             // Confirm connected
-        clientConnectFuture.get(30,TimeUnit.SECONDS);
+            clientConnectFuture.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT);
-        clientSocket.waitForConnected(30,TimeUnit.SECONDS);
+            clientSocket.waitForConnected();
 
             // Have server write slowly.
             int messageCount = 1000;
@@ -143,8 +162,8 @@ public class SlowServerTest
             // Verify receive
             Assert.assertThat("Message Receive Count", clientSocket.messageQueue.size(), is(messageCount));
 
-        // Close
+            // Close server connection (by exiting try-with-resources)
-        serverConn.close(StatusCode.NORMAL);
+        }
 
         Assert.assertTrue("Client Socket Closed", clientSocket.closeLatch.await(10, TimeUnit.SECONDS));
         clientSocket.assertCloseCode(StatusCode.NORMAL);

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/TomcatServerQuirksTest.java

@@ -21,15 +21,19 @@ package org.eclipse.jetty.websocket.client;
 import java.net.URI;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.WebSocketAdapter;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
+import org.eclipse.jetty.websocket.common.test.Timeouts;
+import org.junit.AfterClass;
 import org.junit.Assert;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class TomcatServerQuirksTest
@@ -59,6 +63,21 @@ public class TomcatServerQuirksTest
         }
     }
 
+    private static BlockheadServer server;
+
+    @BeforeClass
+    public static void startServer() throws Exception
+    {
+        server = new BlockheadServer();
+        server.start();
+    }
+
+    @AfterClass
+    public static void stopServer() throws Exception
+    {
+        server.stop();
+    }
+
     /**
      * Test for when encountering a "Transfer-Encoding: chunked" on a Upgrade Response header.
      * <ul>
@@ -71,37 +90,36 @@ public class TomcatServerQuirksTest
     @Test
     public void testTomcat7_0_32_WithTransferEncoding() throws Exception
     {
-        BlockheadServer server = new BlockheadServer();
         WebSocketClient client = new WebSocketClient();
 
+        server.setRequestHandling((req, resp) -> {
+            // Add the extra problematic header that triggers bug found in jetty-io
+            resp.setHeader("Transfer-Encoding", "chunked");
+            return true;
+        });
+
         try
         {
             final int bufferSize = 512;
 
-            server.start();
-
             // Setup Client Factory
             client.start();
 
             // Create End User WebSocket Class
             LatchedSocket websocket = new LatchedSocket();
 
+            CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+            server.addConnectFuture(serverConnFut);
+
             // Open connection
             URI wsURI = server.getWsUri();
             client.connect(websocket,wsURI);
 
-            // Accept incoming connection
-            IBlockheadServerConnection socket = server.accept();
-            socket.setSoTimeout(2000); // timeout
-
-            // Issue upgrade
-            // Add the extra problematic header that triggers bug found in jetty-io
-            socket.addResponseHeader("Transfer-Encoding","chunked");
-            socket.upgrade();
-
             // Wait for proper upgrade
             Assert.assertTrue("Timed out waiting for Client side WebSocket open event",websocket.openLatch.await(1,TimeUnit.SECONDS));
 
+            try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
+            {
                 // Have server write frame.
                 byte payload[] = new byte[bufferSize / 2];
                 Arrays.fill(payload, (byte) 'x');
@@ -113,15 +131,14 @@ public class TomcatServerQuirksTest
                 serverFrame.put((byte) (payload.length & 0xFF)); // second length byte
                 serverFrame.put(payload);
                 BufferUtil.flipToFlush(serverFrame, 0);
-            socket.write(serverFrame);
+                serverConn.writeRaw(serverFrame);
-            socket.flush();
+            }
 
             Assert.assertTrue(websocket.dataLatch.await(1000,TimeUnit.SECONDS));
         }
         finally
         {
             client.stop();
-            server.stop();
         }
     }
 }

jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/WebSocketClientTest.java

@@ -31,6 +31,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -39,35 +40,48 @@ import org.eclipse.jetty.websocket.api.BatchMode;
 import org.eclipse.jetty.websocket.api.RemoteEndpoint;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.UpgradeRequest;
+import org.eclipse.jetty.websocket.common.WebSocketFrame;
 import org.eclipse.jetty.websocket.common.WebSocketSession;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
 import org.eclipse.jetty.websocket.common.io.FutureWriteCallback;
+import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
 import org.eclipse.jetty.websocket.common.test.BlockheadServer;
-import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
 import org.eclipse.jetty.websocket.common.test.Timeouts;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class WebSocketClientTest
 {
-    private BlockheadServer server;
+    private static BlockheadServer server;
     private WebSocketClient client;
 
     @Before
-    public void startClientServer() throws Exception
+    public void startClient() throws Exception
     {
         client = new WebSocketClient();
         client.start();
+    }
+
+    @BeforeClass
+    public static void startServer() throws Exception
+    {
         server = new BlockheadServer();
         server.start();
     }
 
     @After
-    public void stopClientServer() throws Exception
+    public void stopClient() throws Exception
     {
         client.stop();
+    }
+
+    @AfterClass
+    public void stopServer() throws Exception
+    {
         server.stop();
     }
 
@@ -94,13 +108,22 @@ public class WebSocketClientTest
 
         client.getPolicy().setIdleTimeout(10000);
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         ClientUpgradeRequest request = new ClientUpgradeRequest();
         request.setSubProtocols("echo");
         Future<Session> future = client.connect(cliSock,wsUri,request);
 
-        final IBlockheadServerConnection srvSock = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        srvSock.upgrade();
+        {
+            // Setup echo of frames on server side
+            serverConn.setIncomingFrameConsumer((frame)->{
+                WebSocketFrame copy = WebSocketFrame.copy(frame);
+                serverConn.write(copy);
+            });
 
             Session sess = future.get(30,TimeUnit.SECONDS);
             assertThat("Session",sess,notNullValue());
@@ -119,20 +142,10 @@ public class WebSocketClientTest
             if (remote.getBatchMode() == BatchMode.ON)
                 remote.flush();
 
-        try
-        {
-            srvSock.enableIncomingEcho(true);
-            srvSock.startReadThread();
-
             // wait for response from server
             String received = cliSock.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
             assertThat("Message", received, containsString("Hello World"));
         }
-        finally
-        {
-            srvSock.close();
-            srvSock.disconnect();
-        }
     }
 
     @Test
@@ -141,14 +154,17 @@ public class WebSocketClientTest
         client.setMaxIdleTimeout(160000);
         JettyTrackingSocket cliSock = new JettyTrackingSocket();
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         ClientUpgradeRequest request = new ClientUpgradeRequest();
         request.setSubProtocols("echo");
         Future<Session> future = client.connect(cliSock,wsUri,request);
 
-        final IBlockheadServerConnection srvSock = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        srvSock.upgrade();
+        {
-
             Session sess = future.get(30, TimeUnit.SECONDS);
             assertThat("Session", sess, notNullValue());
             assertThat("Session.open", sess.isOpen(), is(true));
@@ -166,17 +182,20 @@ public class WebSocketClientTest
             cliSock.getSession().getRemote().sendString("Hello World!", callback);
             callback.get(1, TimeUnit.SECONDS);
         }
+    }
 
     @Test
     public void testBasicEcho_FromServer() throws Exception
     {
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         JettyTrackingSocket wsocket = new JettyTrackingSocket();
         Future<Session> future = client.connect(wsocket,server.getWsUri());
 
-        // Server
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        final IBlockheadServerConnection srvSock = server.accept();
+        {
-        srvSock.upgrade();
-
             // Validate connect
             Session sess = future.get(30, TimeUnit.SECONDS);
             assertThat("Session", sess, notNullValue());
@@ -185,7 +204,7 @@ public class WebSocketClientTest
             assertThat("Session.upgradeResponse", sess.getUpgradeResponse(), notNullValue());
 
             // Have server send initial message
-        srvSock.write(new TextFrame().setPayload("Hello World"));
+            serverConn.write(new TextFrame().setPayload("Hello World"));
 
             // Verify connect
             future.get(30, TimeUnit.SECONDS);
@@ -194,6 +213,7 @@ public class WebSocketClientTest
             String received = wsocket.messageQueue.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
             assertThat("Message", received, containsString("Hello World"));
         }
+    }
 
     @Test
     public void testLocalRemoteAddress() throws Exception
@@ -203,9 +223,6 @@ public class WebSocketClientTest
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection ssocket = server.accept();
-        ssocket.upgrade();
-
         future.get(30,TimeUnit.SECONDS);
 
         Assert.assertTrue(wsocket.openLatch.await(1,TimeUnit.SECONDS));
@@ -224,37 +241,6 @@ public class WebSocketClientTest
         assertThat("Remote Socket Address / Port",remote.getPort(),greaterThan(0));
     }
 
-    @Test
-    public void testMessageBiggerThanBufferSize() throws Exception
-    {
-        int bufferSize = 512;
-
-        JettyTrackingSocket wsocket = new JettyTrackingSocket();
-
-        URI wsUri = server.getWsUri();
-        Future<Session> future = client.connect(wsocket,wsUri);
-
-        IBlockheadServerConnection ssocket = server.accept();
-        ssocket.upgrade();
-
-        future.get(30,TimeUnit.SECONDS);
-
-        Assert.assertTrue(wsocket.openLatch.await(1,TimeUnit.SECONDS));
-
-        int length = bufferSize + (bufferSize / 2); // 1.5 times buffer size
-        ssocket.write(0x80 | 0x01); // FIN + TEXT
-        ssocket.write(0x7E); // No MASK and 2 bytes length
-        ssocket.write(length >> 8); // first length byte
-        ssocket.write(length & 0xFF); // second length byte
-        for (int i = 0; i < length; ++i)
-        {
-            ssocket.write('x');
-        }
-        ssocket.flush();
-
-        Assert.assertTrue(wsocket.dataLatch.await(1000,TimeUnit.SECONDS));
-    }
-
     /**
      * Ensure that <code>@WebSocket(maxTextMessageSize = 100*1024)</code> behaves as expected.
      * 
@@ -266,11 +252,20 @@ public class WebSocketClientTest
     {
         MaxMessageSocket wsocket = new MaxMessageSocket();
 
+        // Hook into server connection creation
+        CompletableFuture<BlockheadConnection> serverConnFut = new CompletableFuture<>();
+        server.addConnectFuture(serverConnFut);
+
         URI wsUri = server.getWsUri();
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection ssocket = server.accept();
+        try (BlockheadConnection serverConn = serverConnFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
-        ssocket.upgrade();
+        {
+            // Setup echo of frames on server side
+            serverConn.setIncomingFrameConsumer((frame)->{
+                WebSocketFrame copy = WebSocketFrame.copy(frame);
+                serverConn.write(copy);
+            });
 
             wsocket.awaitConnect(1,TimeUnit.SECONDS);
 
@@ -285,10 +280,6 @@ public class WebSocketClientTest
             String msg = StringUtil.toUTF8String(buf,0,buf.length);
 
             wsocket.getSession().getRemote().sendStringByFuture(msg);
-        try
-        {
-            ssocket.enableIncomingEcho(true);
-            ssocket.startReadThread();
 
             // wait for response from server
             wsocket.waitForMessage(1, TimeUnit.SECONDS);
@@ -297,11 +288,6 @@ public class WebSocketClientTest
 
             Assert.assertTrue(wsocket.dataLatch.await(2, TimeUnit.SECONDS));
         }
-        finally
-        {
-            ssocket.close();
-            ssocket.disconnect();
-        }
     }
 
     @Test
@@ -312,9 +298,6 @@ public class WebSocketClientTest
         URI wsUri = server.getWsUri().resolve("/test?snack=cashews&amount=handful&brand=off");
         Future<Session> future = client.connect(wsocket,wsUri);
 
-        IBlockheadServerConnection ssocket = server.accept();
-        ssocket.upgrade();
-
         future.get(30,TimeUnit.SECONDS);
 
         Assert.assertTrue(wsocket.openLatch.await(1,TimeUnit.SECONDS));

jetty-websocket/websocket-common/pom.xml

@@ -36,6 +36,12 @@
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-server</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.eclipse.jetty.toolchain</groupId>
       <artifactId>jetty-test-helper</artifactId>

jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/BlockheadClientRequest.java

@@ -227,6 +227,7 @@ public class BlockheadClientRequest extends HttpRequest implements Response.Comp
 
         endp.setIdleTimeout(client.getIdleTimeout());
 
+        connection.setUpgradeRequestHeaders(this.getHeaders());
         connection.setUpgradeResponseHeaders(response.getHeaders());
 
         // Now swap out the connection

jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/BlockheadConnection.java

@@ -23,6 +23,7 @@ import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.concurrent.Executor;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Consumer;
 
 import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.io.AbstractConnection;
@@ -58,6 +59,7 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
     private final IncomingCapture incomingCapture;
     private ByteBuffer networkBuffer;
     private HttpFields upgradeResponseHeaders;
+    private HttpFields upgradeRequestHeaders;
 
     public BlockheadConnection(WebSocketPolicy policy, ByteBufferPool bufferPool, ExtensionStack extensionStack, EndPoint endp, Executor executor)
     {
@@ -115,6 +117,11 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
         }
     }
 
+    public ByteBufferPool getBufferPool()
+    {
+        return bufferPool;
+    }
+
     public LinkedBlockingQueue<WebSocketFrame> getFrameQueue()
     {
         return incomingCapture.incomingFrames;
@@ -140,6 +147,11 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
         return getEndPoint().getRemoteAddress();
     }
 
+    public HttpFields getUpgradeRequestHeaders()
+    {
+        return upgradeRequestHeaders;
+    }
+
     public HttpFields getUpgradeResponseHeaders()
     {
         return upgradeResponseHeaders;
@@ -175,11 +187,21 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
         LOG.warn("Connection Error", cause);
     }
 
+    public void setUpgradeRequestHeaders(HttpFields upgradeRequestHeaders)
+    {
+        this.upgradeRequestHeaders = upgradeRequestHeaders;
+    }
+
     public void setUpgradeResponseHeaders(HttpFields upgradeResponseHeaders)
     {
         this.upgradeResponseHeaders = upgradeResponseHeaders;
     }
 
+    public void setIncomingFrameConsumer(Consumer<Frame> consumer)
+    {
+        this.incomingCapture.frameConsumer = consumer;
+    }
+
     public void write(WebSocketFrame frame)
     {
         networkOutgoing.outgoingFrame(frame, null, BatchMode.OFF);
@@ -197,7 +219,11 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
         buf.limit(len);
         try
         {
-            getEndPoint().flush(slice);
+            boolean done = false;
+            while (!done)
+            {
+                done = getEndPoint().flush(slice);
+            }
         }
         catch (IOException e)
         {
@@ -311,6 +337,7 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
     {
         public final LinkedBlockingQueue<WebSocketFrame> incomingFrames = new LinkedBlockingQueue<>();
         public final LinkedBlockingQueue<Throwable> incomingErrors = new LinkedBlockingQueue<>();
+        public Consumer<Frame> frameConsumer;
 
         @Override
         public void incomingError(Throwable cause)
@@ -321,6 +348,9 @@ public class BlockheadConnection extends AbstractConnection implements Connectio
         @Override
         public void incomingFrame(Frame frame)
         {
+            if(frameConsumer != null)
+                frameConsumer.accept(frame);
+
             incomingFrames.offer(WebSocketFrame.copy(frame));
         }
     }

jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/BlockheadServer.java

@@ -18,45 +18,116 @@
 
 package org.eclipse.jetty.websocket.common.test;
 
-import static org.hamcrest.Matchers.*;
-
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.ServerSocket;
-import java.net.Socket;
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.BiFunction;
 
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.io.ByteBufferPool;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.HttpConnection;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.Response;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import org.eclipse.jetty.server.handler.DefaultHandler;
+import org.eclipse.jetty.server.handler.HandlerList;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
-import org.junit.Assert;
+import org.eclipse.jetty.websocket.api.WebSocketBehavior;
+import org.eclipse.jetty.websocket.api.WebSocketPolicy;
+import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
+import org.eclipse.jetty.websocket.common.AcceptHash;
+import org.eclipse.jetty.websocket.common.extensions.ExtensionStack;
+import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
+import org.eclipse.jetty.websocket.common.scopes.SimpleContainerScope;
+import org.eclipse.jetty.websocket.common.scopes.WebSocketContainerScope;
 
 /**
- * A overly simplistic websocket server used during testing.
+ * A Server capable of WebSocket upgrade useful for testing.
  * <p>
- * This is not meant to be performant or accurate. In fact, having the server misbehave is a useful trait during testing.
+ * This implementation exists to allow for testing of non-standard server behaviors,
+ * especially around the WebSocket Upgrade process.
  */
 public class BlockheadServer
 {
     private static final Logger LOG = Log.getLogger(BlockheadServer.class);
-    private ServerSocket serverSocket;
+    public static final String SEC_WEBSOCKET_EXTENSIONS = HttpHeader.SEC_WEBSOCKET_EXTENSIONS.toString();
+
+    private final Server server;
+    private final ServerConnector connector;
+    private final BlockheadServerHandler serverHandler;
+    private final WebSocketPolicy policy;
+    private final WebSocketContainerScope websocketContainer;
+    private final WebSocketExtensionFactory extensionFactory;
     private URI wsUri;
 
-    public IBlockheadServerConnection accept() throws IOException
+    public BlockheadServer()
     {
-        LOG.debug(".accept()");
+        this.server = new Server();
-        assertIsStarted();
+        this.connector = new ServerConnector(this.server);
-        Socket socket = serverSocket.accept();
+        this.connector.setPort(0);
-        return new BlockheadServerConnection(socket);
+        this.server.addConnector(connector);
+
+        this.policy = new WebSocketPolicy(WebSocketBehavior.SERVER);
+        this.websocketContainer = new SimpleContainerScope(policy);
+        this.extensionFactory = new WebSocketExtensionFactory(websocketContainer);
+
+        HandlerList handlers = new HandlerList();
+        this.serverHandler = new BlockheadServerHandler(websocketContainer, extensionFactory);
+        handlers.addHandler(this.serverHandler);
+        handlers.addHandler(new DefaultHandler());
+        this.server.setHandler(handlers);
     }
 
-    private void assertIsStarted()
+    public void addConnectFuture(CompletableFuture<BlockheadConnection> serverConnFut)
     {
-        Assert.assertThat("ServerSocket",serverSocket,notNullValue());
+        this.serverHandler.getWSConnectionFutures().offer(serverConnFut);
-        Assert.assertThat("ServerSocket.isBound",serverSocket.isBound(),is(true));
+    }
-        Assert.assertThat("ServerSocket.isClosed",serverSocket.isClosed(),is(false));
 
-        Assert.assertThat("WsUri",wsUri,notNullValue());
+    public WebSocketExtensionFactory getExtensionFactory()
+    {
+        return extensionFactory;
+    }
+
+    public WebSocketPolicy getPolicy()
+    {
+        return policy;
+    }
+
+    public WebSocketContainerScope getWebsocketContainer()
+    {
+        return websocketContainer;
+    }
+
+    /**
+     * Set PRE-Request Handling function.
+     *
+     * @param requestFunction the function to handle the request (before upgrade), do whatever you want.
+     *      Note that if you return true, the request will not process into the default Upgrade flow,
+     *      false will allow the default Upgrade flow.
+     */
+    public void setRequestHandling(BiFunction<Request, Response, Boolean> requestFunction)
+    {
+        this.serverHandler.setFunction(requestFunction);
+    }
+
+    public void resetRequestHandling()
+    {
+        this.serverHandler.setFunction(null);
     }
 
     public URI getWsUri()
@@ -64,28 +135,145 @@ public class BlockheadServer
         return wsUri;
     }
 
-    public void start() throws IOException
+    public void start() throws Exception
     {
-        InetAddress addr = InetAddress.getByName("localhost");
+        this.server.start();
-        serverSocket = new ServerSocket();
+
-        InetSocketAddress endpoint = new InetSocketAddress(addr,0);
+        wsUri = URI.create("ws://localhost:" + this.connector.getLocalPort() + "/");
-        serverSocket.bind(endpoint,1);
+
-        int port = serverSocket.getLocalPort();
+        LOG.debug("BlockheadServer available on {}", wsUri);
-        String uri = String.format("ws://%s:%d/",addr.getHostAddress(),port);
-        wsUri = URI.create(uri);
-        LOG.debug("Server Started on {} -> {}",endpoint,wsUri);
     }
 
-    public void stop()
+    public void stop() throws Exception
     {
         LOG.debug("Stopping Server");
+        this.server.stop();
+    }
+
+    public static class BlockheadServerHandler extends AbstractHandler
+    {
+        private final WebSocketContainerScope container;
+        private final WebSocketExtensionFactory extensionFactory;
+        private BiFunction<Request, Response, Boolean> requestFunction;
+        private LinkedBlockingQueue<CompletableFuture<BlockheadConnection>> futuresQueue;
+
+        public BlockheadServerHandler(WebSocketContainerScope websocketContainer, WebSocketExtensionFactory extensionFactory)
+        {
+            super();
+            this.container = websocketContainer;
+            this.extensionFactory = extensionFactory;
+            this.futuresQueue = new LinkedBlockingQueue<>();
+        }
+
+        public Queue<CompletableFuture<BlockheadConnection>> getWSConnectionFutures()
+        {
+            return futuresQueue;
+        }
+
+        @Override
+        public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
+        {
+            Response baseResponse = (Response) response;
+            if(requestFunction != null)
+            {
+                if(requestFunction.apply(baseRequest, baseResponse))
+                {
+                    baseRequest.setHandled(true);
+                    return;
+                }
+            }
+
+            CompletableFuture<BlockheadConnection> connFut = this.futuresQueue.poll();
+
             try
             {
-            serverSocket.close();
+                baseRequest.setHandled(true);
-        }
+
-        catch (IOException ignore)
+                // default/simplified Upgrade flow
+                String key = request.getHeader("Sec-WebSocket-Key");
+
+                if (key == null)
                 {
-            /* ignore */
+                    throw new IllegalStateException("Missing request header 'Sec-WebSocket-Key'");
+                }
+
+                // build response
+                response.setHeader("Upgrade", "WebSocket");
+                response.addHeader("Connection", "Upgrade");
+                response.addHeader("Sec-WebSocket-Accept", AcceptHash.hashKey(key));
+
+                response.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
+
+                // Initialize / Negotiate Extensions
+                ExtensionStack extensionStack = new ExtensionStack(extensionFactory);
+
+                if (response.containsHeader(SEC_WEBSOCKET_EXTENSIONS))
+                {
+                    // Use pre-negotiated extension list from response
+                    List<ExtensionConfig> extensionConfigs = new ArrayList<>();
+                    response.getHeaders(SEC_WEBSOCKET_EXTENSIONS).forEach(
+                            (value) -> extensionConfigs.addAll(ExtensionConfig.parseList(value)));
+                    extensionStack.negotiate(extensionConfigs);
+                }
+                else
+                {
+                    // Use what was given to us
+                    Enumeration<String> e = request.getHeaders(SEC_WEBSOCKET_EXTENSIONS);
+                    List<ExtensionConfig> extensionConfigs = ExtensionConfig.parseEnum(e);
+                    extensionStack.negotiate(extensionConfigs);
+
+                    String negotiatedHeaderValue = ExtensionConfig.toHeaderValue(extensionStack.getNegotiatedExtensions());
+                    response.setHeader(SEC_WEBSOCKET_EXTENSIONS, negotiatedHeaderValue);
+                }
+
+                WebSocketPolicy policy = this.container.getPolicy().clonePolicy();
+
+                // Get original HTTP connection
+                HttpConnection http = (HttpConnection) request.getAttribute("org.eclipse.jetty.server.HttpConnection");
+
+                EndPoint endp = http.getEndPoint();
+                Connector connector = http.getConnector();
+                Executor executor = connector.getExecutor();
+                ByteBufferPool bufferPool = connector.getByteBufferPool();
+
+                // Setup websocket connection
+                BlockheadServerConnection wsConnection = new BlockheadServerConnection(
+                        policy,
+                        bufferPool,
+                        extensionStack,
+                        endp,
+                        executor);
+
+                if (LOG.isDebugEnabled())
+                {
+                    LOG.debug("HttpConnection: {}", http);
+                    LOG.debug("BlockheadServerConnection: {}", wsConnection);
+                }
+
+                wsConnection.setUpgradeRequestHeaders(baseRequest.getHttpFields());
+                wsConnection.setUpgradeResponseHeaders(baseResponse.getHttpFields());
+
+                // Tell jetty about the new upgraded connection
+                request.setAttribute(HttpConnection.UPGRADE_CONNECTION_ATTRIBUTE, wsConnection);
+
+                if (LOG.isDebugEnabled())
+                    LOG.debug("Websocket upgrade {} {}", request.getRequestURI(), wsConnection);
+
+                if(connFut != null)
+                    connFut.complete(wsConnection);
+
+            }
+            catch(Throwable cause)
+            {
+                if(connFut != null)
+                    connFut.completeExceptionally(cause);
+                LOG.warn(cause);
+            }
+        }
+
+        public void setFunction(BiFunction<Request, Response, Boolean> function)
+        {
+            this.requestFunction = function;
         }
     }
 }

jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/BlockheadServerConnection.java

@@ -18,561 +18,17 @@
 
 package org.eclipse.jetty.websocket.common.test;
 
-import static org.hamcrest.Matchers.is;
+import java.util.concurrent.Executor;
-import static org.hamcrest.Matchers.notNullValue;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.OutputStream;
-import java.net.Socket;
-import java.net.SocketException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import org.eclipse.jetty.io.ByteBufferPool;
-import org.eclipse.jetty.io.MappedByteBufferPool;
+import org.eclipse.jetty.io.EndPoint;
-import org.eclipse.jetty.util.BufferUtil;
-import org.eclipse.jetty.util.IO;
-import org.eclipse.jetty.util.log.Log;
-import org.eclipse.jetty.util.log.Logger;
-import org.eclipse.jetty.websocket.api.BatchMode;
 import org.eclipse.jetty.websocket.api.WebSocketPolicy;
-import org.eclipse.jetty.websocket.api.WriteCallback;
-import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
-import org.eclipse.jetty.websocket.api.extensions.Frame;
-import org.eclipse.jetty.websocket.api.extensions.Frame.Type;
-import org.eclipse.jetty.websocket.api.extensions.IncomingFrames;
-import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
-import org.eclipse.jetty.websocket.common.AcceptHash;
-import org.eclipse.jetty.websocket.common.CloseInfo;
-import org.eclipse.jetty.websocket.common.Generator;
-import org.eclipse.jetty.websocket.common.OpCode;
-import org.eclipse.jetty.websocket.common.Parser;
-import org.eclipse.jetty.websocket.common.WebSocketFrame;
 import org.eclipse.jetty.websocket.common.extensions.ExtensionStack;
-import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
-import org.eclipse.jetty.websocket.common.frames.CloseFrame;
-import org.eclipse.jetty.websocket.common.scopes.SimpleContainerScope;
-import org.junit.Assert;
 
-public class BlockheadServerConnection implements IncomingFrames, OutgoingFrames, Runnable, IBlockheadServerConnection
+public class BlockheadServerConnection extends BlockheadConnection
 {
-    private static final Logger LOG = Log.getLogger(BlockheadServerConnection.class);
+    public BlockheadServerConnection(WebSocketPolicy policy, ByteBufferPool bufferPool, ExtensionStack extensionStack, EndPoint endp, Executor executor)
-    
-    private final int BUFFER_SIZE = 8192;
-    private final Socket socket;
-    private final ByteBufferPool bufferPool;
-    private final WebSocketPolicy policy;
-    private final LinkedBlockingQueue<WebSocketFrame> incomingFrames = new LinkedBlockingQueue<>();
-    private final LinkedBlockingQueue<Throwable> incomingErrors = new LinkedBlockingQueue<>();
-    private final Parser parser;
-    private final Generator generator;
-    private final AtomicInteger parseCount;
-    private final WebSocketExtensionFactory extensionRegistry;
-    private final AtomicBoolean echoing = new AtomicBoolean(false);
-    private final AtomicBoolean reading = new AtomicBoolean(false);
-    private Thread readThread;
-
-    /** Set to true to disable timeouts (for debugging reasons) */
-    private boolean debug = false;
-    private OutputStream out;
-    private InputStream in;
-
-    private Map<String, String> extraResponseHeaders = new HashMap<>();
-    private OutgoingFrames outgoing = this;
-
-    public BlockheadServerConnection(Socket socket)
     {
-        this.socket = socket;
+        super(policy, bufferPool, extensionStack, endp, executor);
-        this.policy = WebSocketPolicy.newServerPolicy();
-        this.policy.setMaxBinaryMessageSize(100000);
-        this.policy.setMaxTextMessageSize(100000);
-        // This is a blockhead server connection, no point tracking leaks on this object.
-        this.bufferPool = new MappedByteBufferPool(BUFFER_SIZE);
-        this.parser = new Parser(policy,bufferPool);
-        this.parseCount = new AtomicInteger(0);
-        this.generator = new Generator(policy,bufferPool,false);
-        this.extensionRegistry = new WebSocketExtensionFactory(new SimpleContainerScope(policy,bufferPool));
-    }
-
-    /**
-     * Add an extra header for the upgrade response (from the server). No extra work is done to ensure the key and value are sane for http.
-     * @param rawkey the raw key
-     * @param rawvalue the raw value
-     */
-    @Override
-    public void addResponseHeader(String rawkey, String rawvalue)
-    {
-        extraResponseHeaders.put(rawkey,rawvalue);
-    }
-
-    /* (non-Javadoc)
-     * @see org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection#close()
-     */
-    @Override
-    public void close() throws IOException
-    {
-        write(new CloseFrame());
-        flush();
-    }
-
-    /* (non-Javadoc)
-     * @see org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection#close(int)
-     */
-    @Override
-    public void close(int statusCode) throws IOException
-    {
-        CloseInfo close = new CloseInfo(statusCode);
-        write(close.asFrame());
-        flush();
-    }
-
-    @Override
-    public void disconnect()
-    {
-        LOG.debug("disconnect");
-        reading.set(false);
-        IO.close(in);
-        IO.close(out);
-        if (socket != null)
-        {
-            try
-            {
-                socket.close();
-            }
-            catch (IOException ignore)
-            {
-                /* ignore */
-            }
-        }
-    }
-    @Override
-    public void flush() throws IOException
-    {
-        getOutputStream().flush();
-    }
-
-    @Override
-    public ByteBufferPool getBufferPool()
-    {
-        return bufferPool;
-    }
-
-    public InputStream getInputStream() throws IOException
-    {
-        if (in == null)
-        {
-            in = socket.getInputStream();
-        }
-        return in;
-    }
-
-    private OutputStream getOutputStream() throws IOException
-    {
-        if (out == null)
-        {
-            out = socket.getOutputStream();
-        }
-        return out;
-    }
-
-    @Override
-    public Parser getParser()
-    {
-        return parser;
-    }
-
-    public WebSocketPolicy getPolicy()
-    {
-        return policy;
-    }
-
-    @Override
-    public void incomingError(Throwable e)
-    {
-        incomingErrors.offer(e);
-    }
-
-    @Override
-    public void incomingFrame(Frame frame)
-    {
-        LOG.debug("incomingFrame({})",frame);
-        int count = parseCount.incrementAndGet();
-        if ((count % 10) == 0)
-        {
-            LOG.info("Server parsed {} frames",count);
-        }
-        incomingFrames.offer(WebSocketFrame.copy(frame));
-
-        if (frame.getOpCode() == OpCode.CLOSE)
-        {
-            CloseInfo close = new CloseInfo(frame);
-            LOG.debug("Close frame: {}",close);
-        }
-
-        Type type = frame.getType();
-
-        if (echoing.get() && (type.isData() || type.isContinuation()))
-        {
-            try
-            {
-                write(WebSocketFrame.copy(frame).setMasked(false));
-            }
-            catch (IOException e)
-            {
-                LOG.warn(e);
-            }
-        }
-    }
-
-    @Override
-    public void outgoingFrame(Frame frame, WriteCallback callback, BatchMode batchMode)
-    {
-        ByteBuffer headerBuf = generator.generateHeaderBytes(frame);
-        if (LOG.isDebugEnabled())
-        {
-            LOG.debug("writing out: {}",BufferUtil.toDetailString(headerBuf));
-        }
-
-        try
-        {
-            BufferUtil.writeTo(headerBuf,out);
-            if (frame.hasPayload())
-                BufferUtil.writeTo(frame.getPayload(),out);
-            out.flush();
-            if (callback != null)
-            {
-                callback.writeSuccess();
-            }
-
-            if (frame.getOpCode() == OpCode.CLOSE)
-            {
-                disconnect();
-            }
-        }
-        catch (Throwable t)
-        {
-            if (callback != null)
-            {
-                callback.writeFailed(t);
-            }
-        }
-    }
-
-    public List<ExtensionConfig> parseExtensions(List<String> requestLines)
-    {
-        List<ExtensionConfig> extensionConfigs = new ArrayList<>();
-        
-        List<String> hits = regexFind(requestLines, "^Sec-WebSocket-Extensions: (.*)$");
-
-        for (String econf : hits)
-        {
-            // found extensions
-            ExtensionConfig config = ExtensionConfig.parse(econf);
-            extensionConfigs.add(config);
-        }
-
-        return extensionConfigs;
-    }
-
-    @Override
-    public String parseWebSocketKey(List<String> requestLines)
-    {
-        List<String> hits = regexFind(requestLines,"^Sec-WebSocket-Key: (.*)$");
-        if (hits.size() <= 0)
-        {
-            return null;
-        }
-        
-        Assert.assertThat("Number of Sec-WebSocket-Key headers", hits.size(), is(1));
-        
-        String key = hits.get(0);
-        return key;
-    }
-
-    @Override
-    public int read(ByteBuffer buf) throws IOException
-    {
-        int len = 0;
-        while ((in.available() > 0) && (buf.remaining() > 0))
-        {
-            buf.put((byte)in.read());
-            len++;
-        }
-        return len;
-    }
-
-    @Override
-    public LinkedBlockingQueue<WebSocketFrame> getFrameQueue()
-    {
-        return incomingFrames;
-    }
-
-    @Override
-    public String readRequest() throws IOException
-    {
-        LOG.debug("Reading client request");
-        StringBuilder request = new StringBuilder();
-        BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
-        for (String line = in.readLine(); line != null; line = in.readLine())
-        {
-            if (line.length() == 0)
-            {
-                break;
-            }
-            request.append(line).append("\r\n");
-            LOG.debug("read line: {}",line);
-        }
-
-        LOG.debug("Client Request:{}{}","\n",request);
-        return request.toString();
-    }
-
-    @Override
-    public List<String> readRequestLines() throws IOException
-    {
-        LOG.debug("Reading client request header");
-        List<String> lines = new ArrayList<>();
-
-        BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
-        for (String line = in.readLine(); line != null; line = in.readLine())
-        {
-            if (line.length() == 0)
-            {
-                break;
-            }
-            lines.add(line);
-        }
-
-        return lines;
-    }
-
-    @Override
-    public List<String> regexFind(List<String> lines, String pattern)
-    {
-        List<String> hits = new ArrayList<>();
-
-        Pattern patKey = Pattern.compile(pattern,Pattern.CASE_INSENSITIVE);
-
-        Matcher mat;
-        for (String line : lines)
-        {
-            mat = patKey.matcher(line);
-            if (mat.matches())
-            {
-                if (mat.groupCount() >= 1)
-                {
-                    hits.add(mat.group(1));
-                }
-                else
-                {
-                    hits.add(mat.group(0));
-                }
-            }
-        }
-
-        return hits;
-    }
-
-    @Override
-    public void respond(String rawstr) throws IOException
-    {
-        LOG.debug("respond(){}{}","\n",rawstr);
-        getOutputStream().write(rawstr.getBytes());
-        flush();
-    }
-
-    @Override
-    public void run()
-    {
-        LOG.debug("Entering read thread");
-
-        long totalReadBytes = 0;
-        ByteBuffer buf = bufferPool.acquire(BUFFER_SIZE, false);
-        while(reading.get())
-        {
-            try
-            {
-                BufferUtil.clearToFill(buf);
-                long len = read(buf);
-                if (len > 0)
-                {
-                    totalReadBytes += len;
-                    LOG.debug("Read {} bytes", len);
-                    BufferUtil.flipToFlush(buf, 0);
-                    parser.parse(buf);
-                }
-
-                try
-                {
-                    TimeUnit.MILLISECONDS.sleep(20);
-                }
-                catch (InterruptedException gnore)
-                {
-                    /* ignore */
-                }
-            }
-            catch (IOException e)
-            {
-                LOG.debug("Exception during echo loop", e);
-            }
-            catch (Throwable t)
-            {
-                LOG.warn("Exception during echo loop", t);
-            }
-        }
-
-        LOG.debug("Read {} total bytes (exiting)",totalReadBytes);
-        bufferPool.release(buf);
-    }
-
-    @Override
-    public void setSoTimeout(int ms) throws SocketException
-    {
-        socket.setSoTimeout(ms);
-    }
-
-    @Override
-    public void startReadThread()
-    {
-        if (readThread != null)
-        {
-            throw new IllegalStateException("Read thread already declared/started!");
-        }
-        readThread = new Thread(this,"Blockhead-Server-Read");
-        LOG.debug("Starting Read Thread: {}", readThread);
-        reading.set(true);
-        readThread.start();
-    }
-
-    @Override
-    public void enableIncomingEcho(boolean enabled)
-    {
-        echoing.set(enabled);
-    }
-
-    @Override
-    public List<String> upgrade() throws IOException
-    {
-        List<String> requestLines = readRequestLines();
-        List<ExtensionConfig> extensionConfigs = parseExtensions(requestLines);
-        String key = parseWebSocketKey(requestLines);
-
-        LOG.debug("Client Request Extensions: {}",extensionConfigs);
-        LOG.debug("Client Request Key: {}",key);
-
-        Assert.assertThat("Request: Sec-WebSocket-Key",key,notNullValue());
-
-        // collect extensions configured in response header
-        ExtensionStack extensionStack = new ExtensionStack(extensionRegistry);
-        extensionStack.negotiate(extensionConfigs);
-
-        // Start with default routing
-        extensionStack.setNextIncoming(this);
-        extensionStack.setNextOutgoing(this);
-
-        // Configure Parser / Generator
-        extensionStack.configure(parser);
-        extensionStack.configure(generator);
-
-        // Start Stack
-        try
-        {
-            extensionStack.start();
-        }
-        catch (Exception e)
-        {
-            throw new IOException("Unable to start Extension Stack");
-        }
-
-        // Configure Parser
-        parser.setIncomingFramesHandler(extensionStack);
-
-        // Setup Response
-        StringBuilder resp = new StringBuilder();
-        resp.append("HTTP/1.1 101 Upgrade\r\n");
-        resp.append("Connection: upgrade\r\n");
-        resp.append("Content-Length: 0\r\n");
-        resp.append("Sec-WebSocket-Accept: ");
-        resp.append(AcceptHash.hashKey(key)).append("\r\n");
-        if (extensionStack.hasNegotiatedExtensions())
-        {
-            // Respond to used extensions
-            resp.append("Sec-WebSocket-Extensions: ");
-            boolean delim = false;
-            for (ExtensionConfig ext : extensionStack.getNegotiatedExtensions())
-            {
-                if (delim)
-                {
-                    resp.append(", ");
-                }
-                resp.append(ext.getParameterizedName());
-                delim = true;
-            }
-            resp.append("\r\n");
-        }
-        if (extraResponseHeaders.size() > 0)
-        {
-            for (Map.Entry<String, String> xheader : extraResponseHeaders.entrySet())
-            {
-                resp.append(xheader.getKey());
-                resp.append(": ");
-                resp.append(xheader.getValue());
-                resp.append("\r\n");
-            }
-        }
-        resp.append("\r\n");
-        
-        // Write Response
-        LOG.debug("Response: {}",resp.toString());
-        write(resp.toString().getBytes());
-        return requestLines;
-    }
-
-    private void write(byte[] bytes) throws IOException
-    {
-        getOutputStream().write(bytes);
-    }
-
-    public void write(byte[] buf, int offset, int length) throws IOException
-    {
-        getOutputStream().write(buf,offset,length);
-    }
-
-    /* (non-Javadoc)
-     * @see org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection#write(org.eclipse.jetty.websocket.api.extensions.Frame)
-     */
-    @Override
-    public void write(Frame frame) throws IOException
-    {
-        LOG.debug("write(Frame->{}) to {}",frame,outgoing);
-        outgoing.outgoingFrame(frame,null,BatchMode.OFF);
-    }
-
-    @Override
-    public void write(int b) throws IOException
-    {
-        getOutputStream().write(b);
-    }
-
-    @Override
-    public void write(ByteBuffer buf) throws IOException
-    {
-        byte arr[] = BufferUtil.toArray(buf);
-        if ((arr != null) && (arr.length > 0))
-        {
-            getOutputStream().write(arr);
-        }
     }
 }

jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/IBlockheadServerConnection.java

@@ -1,68 +0,0 @@
-//
-//  ========================================================================
-//  Copyright (c) 1995-2018 Mort Bay Consulting Pty. Ltd.
-//  ------------------------------------------------------------------------
-//  All rights reserved. This program and the accompanying materials
-//  are made available under the terms of the Eclipse Public License v1.0
-//  and Apache License v2.0 which accompanies this distribution.
-//
-//      The Eclipse Public License is available at
-//      http://www.eclipse.org/legal/epl-v10.html
-//
-//      The Apache License v2.0 is available at
-//      http://www.opensource.org/licenses/apache2.0.php
-//
-//  You may elect to redistribute this code under either of these licenses.
-//  ========================================================================
-//
-
-package org.eclipse.jetty.websocket.common.test;
-
-import java.io.IOException;
-import java.net.SocketException;
-import java.nio.ByteBuffer;
-import java.util.List;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.eclipse.jetty.io.ByteBufferPool;
-import org.eclipse.jetty.websocket.api.extensions.Frame;
-import org.eclipse.jetty.websocket.common.Parser;
-import org.eclipse.jetty.websocket.common.WebSocketFrame;
-
-@Deprecated
-public interface IBlockheadServerConnection
-{
-    public void close() throws IOException;
-    public void close(int statusCode) throws IOException;
-    public void disconnect();
-
-    public void write(Frame frame) throws IOException;
-    public void write(ByteBuffer buf) throws IOException;
-    public void write(int b) throws IOException;
-    public void flush() throws IOException;
-
-    public LinkedBlockingQueue<WebSocketFrame> getFrameQueue();
-
-    public void enableIncomingEcho(boolean enabled);
-    public void startReadThread();
-
-    public String readRequest() throws IOException;
-    public List<String> readRequestLines() throws IOException;
-    public String parseWebSocketKey(List<String> requestLines);
-    /**
-     * Add an extra header for the upgrade response (from the server). No extra work is done to ensure the key and value are sane for http.
-     * @param rawkey the raw key
-     * @param rawvalue the raw value
-     */
-    public void addResponseHeader(String rawkey, String rawvalue);
-    public List<String> upgrade() throws IOException;
-    public void setSoTimeout(int ms) throws SocketException;
-
-    public void respond(String rawstr) throws IOException;
-    public List<String> regexFind(List<String> lines, String pattern);
-    public ByteBufferPool getBufferPool();
-    public Parser getParser();
-
-    @Deprecated
-    public int read(ByteBuffer buf) throws IOException;
-}

jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/Timeouts.java

@@ -20,11 +20,20 @@ package org.eclipse.jetty.websocket.common.test;
 
 import java.util.concurrent.TimeUnit;
 
+/**
+ * A central place for all of the various test timeouts within the websocket testing.
+ */
 public class Timeouts
 {
+    // establish a connection timeout
     public static final long CONNECT = 2;
     public static final TimeUnit CONNECT_UNIT = TimeUnit.SECONDS;
 
+    // poll for an event timeout
     public static final long POLL_EVENT = 2;
     public static final TimeUnit POLL_EVENT_UNIT = TimeUnit.SECONDS;
+
+    // send a message timeout
+    public static final long SEND = 2;
+    public static final TimeUnit SEND_UNIT = TimeUnit.SECONDS;
 }

jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/ab/TestABCase9.java

@@ -70,7 +70,6 @@ public class TestABCase9 extends AbstractABCase
         int offset = 0;
         boolean fin;
         ByteBuffer buf;
-        ;
         byte op = opcode;
         while (remaining > 0)
         {