474936 - WebSocketSessions are not always cleaned out from openSessions

+ Adding testcase
+ Enabling Connection.onClose() -> ioState.onDisconnected()

Conflicts:
	jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/WebSocketCloseTest.java

2 files changed, 189 insertions, 13 deletions

diff --git a/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/io/AbstractWebSocketConnection.java b/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/io/AbstractWebSocketConnection.java
index 5f9b230730..c0cdfadba7 100644
--- a/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/io/AbstractWebSocketConnection.java
+++ b/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/io/AbstractWebSocketConnection.java
@@ -412,7 +412,7 @@ public abstract class AbstractWebSocketConnection extends AbstractConnection imp
         if (LOG.isDebugEnabled())
             LOG.debug("{} onClose()",policy.getBehavior());
         super.onClose();
-        // ioState.onDisconnected();
+        ioState.onDisconnected();
         flusher.close();
     }
 
diff --git a/jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/WebSocketCloseTest.java b/jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/WebSocketCloseTest.java
index f5ba27f199..940829def6 100644
--- a/jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/WebSocketCloseTest.java
+++ b/jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/WebSocketCloseTest.java
@@ -19,9 +19,11 @@
 package org.eclipse.jetty.websocket.server;
 
 import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -35,7 +37,9 @@ import org.eclipse.jetty.websocket.api.WebSocketAdapter;
 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.WebSocketSession;
 import org.eclipse.jetty.websocket.common.events.AbstractEventDriver;
+import org.eclipse.jetty.websocket.common.frames.TextFrame;
 import org.eclipse.jetty.websocket.common.test.BlockheadClient;
 import org.eclipse.jetty.websocket.server.helper.RFCSocket;
 import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
@@ -44,7 +48,6 @@ import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
 import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
 import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
 import org.junit.AfterClass;
-import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -79,10 +82,16 @@ public class WebSocketCloseTest
     @SuppressWarnings("serial")
     public static class CloseServlet extends WebSocketServlet implements WebSocketCreator
     {
+        private WebSocketServerFactory serverFactory;
+
         @Override
         public void configure(WebSocketServletFactory factory)
         {
             factory.setCreator(this);
+            if (factory instanceof WebSocketServerFactory)
+            {
+                this.serverFactory = (WebSocketServerFactory)factory;
+            }
         }
 
         @Override
@@ -99,11 +108,59 @@ public class WebSocketCloseTest
                 closeSocket = new FastFailSocket();
                 return closeSocket;
             }
+
+            if (req.hasSubProtocol("container"))
+            {
+                closeSocket = new ContainerSocket(serverFactory);
+                return closeSocket;
+            }
             return new RFCSocket();
         }
     }
 
     /**
+     * On Message, return container information
+     */
+    public static class ContainerSocket extends AbstractCloseSocket
+    {
+        private static final Logger LOG = Log.getLogger(WebSocketCloseTest.ContainerSocket.class);
+        private final WebSocketServerFactory container;
+        private Session session;
+
+        public ContainerSocket(WebSocketServerFactory container)
+        {
+            this.container = container;
+        }
+
+        @Override
+        public void onWebSocketText(String message)
+        {
+            LOG.debug("onWebSocketText({})",message);
+            if (message.equalsIgnoreCase("openSessions"))
+            {
+                Set<WebSocketSession> sessions = container.getOpenSessions();
+
+                StringBuilder ret = new StringBuilder();
+                ret.append("openSessions.size=").append(sessions.size()).append('\n');
+                int idx = 0;
+                for (WebSocketSession sess : sessions)
+                {
+                    ret.append('[').append(idx++).append("] ").append(sess.toString()).append('\n');
+                }
+                session.getRemote().sendStringByFuture(ret.toString());
+            }
+            session.close(StatusCode.NORMAL,"ContainerSocket");
+        }
+
+        @Override
+        public void onWebSocketConnect(Session sess)
+        {
+            LOG.debug("onWebSocketConnect({})",sess);
+            this.session = sess;
+        }
+    }
+
+    /**
      * On Connect, close socket
      */
     public static class FastCloseSocket extends AbstractCloseSocket
@@ -155,6 +212,9 @@ public class WebSocketCloseTest
 
     /**
      * Test fast close (bug #403817)
+     * 
+     * @throws Exception
+     *             on test failure
      */
     @Test
     public void testFastClose() throws Exception
@@ -170,21 +230,24 @@ public class WebSocketCloseTest
             // Verify that client got close frame
             EventQueue<WebSocketFrame> frames = client.readFrames(1,1,TimeUnit.SECONDS);
             WebSocketFrame frame = frames.poll();
-            Assert.assertThat("frames[0].opcode",frame.getOpCode(),is(OpCode.CLOSE));
+            assertThat("frames[0].opcode",frame.getOpCode(),is(OpCode.CLOSE));
             CloseInfo close = new CloseInfo(frame);
-            Assert.assertThat("Close Status Code",close.getStatusCode(),is(StatusCode.NORMAL));
-            
+            assertThat("Close Status Code",close.getStatusCode(),is(StatusCode.NORMAL));
+
             // Notify server of close handshake
             client.write(close.asFrame()); // respond with close
-            
+
             // ensure server socket got close event
-            Assert.assertThat("Fast Close Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
-            Assert.assertThat("Fast Close.statusCode",closeSocket.closeStatusCode,is(StatusCode.NORMAL));
+            assertThat("Fast Close Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
+            assertThat("Fast Close.statusCode",closeSocket.closeStatusCode,is(StatusCode.NORMAL));
         }
     }
 
     /**
      * Test fast fail (bug #410537)
+     * 
+     * @throws Exception
+     *             on test failure
      */
     @Test
     public void testFastFail() throws Exception
@@ -201,16 +264,129 @@ public class WebSocketCloseTest
 
                 EventQueue<WebSocketFrame> frames = client.readFrames(1,1,TimeUnit.SECONDS);
                 WebSocketFrame frame = frames.poll();
-                Assert.assertThat("frames[0].opcode",frame.getOpCode(),is(OpCode.CLOSE));
+                assertThat("frames[0].opcode",frame.getOpCode(),is(OpCode.CLOSE));
                 CloseInfo close = new CloseInfo(frame);
-                Assert.assertThat("Close Status Code",close.getStatusCode(),is(StatusCode.SERVER_ERROR));
+                assertThat("Close Status Code",close.getStatusCode(),is(StatusCode.SERVER_ERROR));
 
                 client.write(close.asFrame()); // respond with close
 
                 // ensure server socket got close event
-                Assert.assertThat("Fast Fail Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
-                Assert.assertThat("Fast Fail.statusCode",closeSocket.closeStatusCode,is(StatusCode.SERVER_ERROR));
-                Assert.assertThat("Fast Fail.errors",closeSocket.errors.size(),is(1));
+                assertThat("Fast Fail Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
+                assertThat("Fast Fail.statusCode",closeSocket.closeStatusCode,is(StatusCode.SERVER_ERROR));
+                assertThat("Fast Fail.errors",closeSocket.errors.size(),is(1));
+            }
+        }
+    }
+
+    /**
+     * Test session open session cleanup (bug #474936)
+     * 
+     * @throws Exception
+     *             on test failure
+     */
+    @Test
+    public void testOpenSessionCleanup() throws Exception
+    {
+        fastFail();
+        fastClose();
+        dropConnection();
+
+        try (BlockheadClient client = new BlockheadClient(server.getServerUri()))
+        {
+            client.setProtocols("container");
+            client.setTimeout(1,TimeUnit.SECONDS);
+            client.connect();
+            client.sendStandardRequest();
+            client.expectUpgradeResponse();
+
+            TextFrame text = new TextFrame();
+            text.setPayload("openSessions");
+            client.write(text);
+
+            EventQueue<WebSocketFrame> frames = client.readFrames(2,1,TimeUnit.SECONDS);
+            WebSocketFrame frame = frames.poll();
+            assertThat("frames[0].opcode",frame.getOpCode(),is(OpCode.TEXT));
+
+            String resp = frame.getPayloadAsUTF8();
+            assertThat("Should only have 1 open session",resp,containsString("openSessions.size=1\n"));
+
+            frame = frames.poll();
+            assertThat("frames[1].opcode",frame.getOpCode(),is(OpCode.CLOSE));
+            CloseInfo close = new CloseInfo(frame);
+            assertThat("Close Status Code",close.getStatusCode(),is(StatusCode.NORMAL));
+            client.write(close.asFrame()); // respond with close
+
+            // ensure server socket got close event
+            assertThat("Open Sessions Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
+            assertThat("Open Sessions.statusCode",closeSocket.closeStatusCode,is(StatusCode.NORMAL));
+            assertThat("Open Sessions.errors",closeSocket.errors.size(),is(0));
+        }
+    }
+
+    private void fastClose() throws Exception
+    {
+        try (BlockheadClient client = new BlockheadClient(server.getServerUri()))
+        {
+            client.setProtocols("fastclose");
+            client.setTimeout(1,TimeUnit.SECONDS);
+            try (StacklessLogging scope = new StacklessLogging(WebSocketSession.class))
+            {
+                client.connect();
+                client.sendStandardRequest();
+                client.expectUpgradeResponse();
+                
+                client.readFrames(1,1,TimeUnit.SECONDS);
+
+                CloseInfo close = new CloseInfo(StatusCode.NORMAL,"Normal");
+                assertThat("Close Status Code",close.getStatusCode(),is(StatusCode.NORMAL));
+
+                // Notify server of close handshake
+                client.write(close.asFrame()); // respond with close
+
+                // ensure server socket got close event
+                assertThat("Fast Close Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
+                assertThat("Fast Close.statusCode",closeSocket.closeStatusCode,is(StatusCode.NORMAL));
+            }
+        }
+    }
+
+    private void fastFail() throws Exception
+    {
+        try (BlockheadClient client = new BlockheadClient(server.getServerUri()))
+        {
+            client.setProtocols("fastfail");
+            client.setTimeout(1,TimeUnit.SECONDS);
+            try (StacklessLogging scope = new StacklessLogging(WebSocketSession.class))
+            {
+                client.connect();
+                client.sendStandardRequest();
+                client.expectUpgradeResponse();
+                
+                client.readFrames(1,1,TimeUnit.SECONDS);
+
+                CloseInfo close = new CloseInfo(StatusCode.NORMAL,"Normal");
+                client.write(close.asFrame()); // respond with close
+
+                // ensure server socket got close event
+                assertThat("Fast Fail Latch",closeSocket.closeLatch.await(1,TimeUnit.SECONDS),is(true));
+                assertThat("Fast Fail.statusCode",closeSocket.closeStatusCode,is(StatusCode.SERVER_ERROR));
+                assertThat("Fast Fail.errors",closeSocket.errors.size(),is(1));
+            }
+        }
+    }
+    
+    private void dropConnection() throws Exception
+    {
+        try (BlockheadClient client = new BlockheadClient(server.getServerUri()))
+        {
+            client.setProtocols("container");
+            client.setTimeout(1,TimeUnit.SECONDS);
+            try (StacklessLogging scope = new StacklessLogging(WebSocketSession.class))
+            {
+                client.connect();
+                client.sendStandardRequest();
+                client.expectUpgradeResponse();
+                client.disconnect();
             }
         }
     }