Merge pull request #2658 from eclipse/jetty-9.4.x-issue-2655-wsclient-session-removal

Issue #2655 - Removing closed WebSocket Session's from WebSocketClient

Author: joakime

jetty-websocket/javax-websocket-server-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/server/DelayedStartClientOnServerTest.java

@@ -247,15 +247,15 @@ public void testHttpClientThreads_AfterClientConnectTo() throws Exception
             assertThat("Response", response, startsWith("Connected to ws://"));
             List<String> threadNames = getThreadNames(server);
             assertNoHttpClientPoolThreads(threadNames);
-            assertThat("Threads", threadNames, hasItem(containsString("WebSocketContainer@")));
+            assertThat("Threads", threadNames, hasItem(containsString("WebSocketClient@")));
         }
         finally
         {
             server.stop();
         }
     }
 
-    @Test
+    @Test(timeout = 5000)
     public void testHttpClientThreads_AfterServerConnectTo() throws Exception
     {
         Server server = new Server(0);

jetty-websocket/websocket-api/src/main/java/org/eclipse/jetty/websocket/api/WebSocketPolicy.java

@@ -144,7 +144,12 @@ public void assertValidTextMessageSize(int requestedSize)
 
     public WebSocketPolicy clonePolicy()
     {
-        WebSocketPolicy clone = new WebSocketPolicy(this.behavior);
+        return clonePolicy(this.behavior);
+    }
+
+    public WebSocketPolicy clonePolicy(WebSocketBehavior behavior)
+    {
+        WebSocketPolicy clone = new WebSocketPolicy(behavior);
         clone.idleTimeout = this.idleTimeout;
         clone.maxTextMessageSize = this.maxTextMessageSize;
         clone.maxTextMessageBufferSize = this.maxTextMessageBufferSize;

jetty-websocket/websocket-client/src/main/java/org/eclipse/jetty/websocket/client/DefaultHttpClientProvider.java

@@ -21,6 +21,8 @@
 import java.util.concurrent.Executor;
 
 import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.io.ByteBufferPool;
+import org.eclipse.jetty.io.MappedByteBufferPool;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
 import org.eclipse.jetty.websocket.common.scopes.WebSocketContainerScope;
@@ -31,11 +33,13 @@ public static HttpClient newHttpClient(WebSocketContainerScope scope)
     {
         SslContextFactory sslContextFactory = null;
         Executor executor = null;
+        ByteBufferPool bufferPool = null;
 
         if (scope != null)
         {
             sslContextFactory = scope.getSslContextFactory();
             executor = scope.getExecutor();
+            bufferPool = scope.getBufferPool();
         }
 
         if (sslContextFactory == null)
@@ -53,6 +57,13 @@ public static HttpClient newHttpClient(WebSocketContainerScope scope)
             executor = threadPool;
         }
         client.setExecutor(executor);
+
+        if (bufferPool == null)
+        {
+            bufferPool = new MappedByteBufferPool();
+        }
+        client.setByteBufferPool(bufferPool);
+
         return client;
     }
 }

jetty-websocket/websocket-client/src/main/java/org/eclipse/jetty/websocket/client/WebSocketClient.java

@@ -29,11 +29,10 @@
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
-import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Supplier;
 
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.io.ByteBufferPool;
-import org.eclipse.jetty.io.MappedByteBufferPool;
 import org.eclipse.jetty.util.DecoratedObjectFactory;
 import org.eclipse.jetty.util.StringUtil;
 import org.eclipse.jetty.util.component.ContainerLifeCycle;
@@ -56,7 +55,6 @@
 import org.eclipse.jetty.websocket.common.WebSocketSessionFactory;
 import org.eclipse.jetty.websocket.common.events.EventDriverFactory;
 import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
-import org.eclipse.jetty.websocket.common.scopes.DelegatedContainerScope;
 import org.eclipse.jetty.websocket.common.scopes.SimpleContainerScope;
 import org.eclipse.jetty.websocket.common.scopes.WebSocketContainerScope;
 
@@ -70,14 +68,15 @@ public class WebSocketClient extends ContainerLifeCycle implements WebSocketCont
     // From HttpClient
     private final HttpClient httpClient;
 
-    //
-    private final WebSocketContainerScope containerScope;
+    // CDI layer
+    private final Supplier<DecoratedObjectFactory> objectFactorySupplier;
+
+    // WebSocket Specifics
+    private final WebSocketPolicy policy;
     private final WebSocketExtensionFactory extensionRegistry;
     private final EventDriverFactory eventDriverFactory;
     private final SessionFactory sessionFactory;
 
-    private final int id = ThreadLocalRandom.current().nextInt();
-
     // defaults to true for backwards compatibility
     private boolean stopAtShutdown = true;
 
@@ -86,9 +85,7 @@ public class WebSocketClient extends ContainerLifeCycle implements WebSocketCont
      */
     public WebSocketClient()
     {
-        // Create synthetic HttpClient
-        this(HttpClientProvider.get(null));
-        addBean(this.httpClient);
+        this((HttpClient)null);
     }
 
     /**
@@ -99,7 +96,7 @@ public WebSocketClient()
      */
     public WebSocketClient(HttpClient httpClient)
     {
-        this(httpClient,new DecoratedObjectFactory());
+        this(httpClient, null);
     }
 
     /**
@@ -112,46 +109,45 @@ public WebSocketClient(HttpClient httpClient)
      */
     public WebSocketClient(HttpClient httpClient, DecoratedObjectFactory objectFactory)
     {
-        this.containerScope = new SimpleContainerScope(WebSocketPolicy.newClientPolicy(),new MappedByteBufferPool(),objectFactory);
-        this.httpClient = httpClient;
-        this.extensionRegistry = new WebSocketExtensionFactory(containerScope);
-        this.eventDriverFactory = new EventDriverFactory(containerScope);
-        this.sessionFactory = new WebSocketSessionFactory(containerScope);
+        this(new SimpleContainerScope(new WebSocketPolicy(WebSocketBehavior.CLIENT), null, null, null, objectFactory), null, null, httpClient);
     }
 
     /**
      * Create a new WebSocketClient
      *
-     * @param executor
-     *            the executor to use
+     * @param sslContextFactory
+     *            ssl context factory to use
      * @deprecated use {@link #WebSocketClient(HttpClient)} instead
      */
     @Deprecated
-    public WebSocketClient(Executor executor)
+    public WebSocketClient(SslContextFactory sslContextFactory)
     {
-        this(null,executor);
+        this(sslContextFactory,null, null);
     }
 
     /**
      * Create a new WebSocketClient
      *
-     * @param bufferPool
-     *            byte buffer pool to use
+     * @param executor
+     *            the executor to use
+     * @deprecated use {@link #WebSocketClient(HttpClient)} instead
      */
-    public WebSocketClient(ByteBufferPool bufferPool)
+    public WebSocketClient(Executor executor)
     {
-        this(null,null,bufferPool);
+        this(null, executor, null);
     }
 
     /**
      * Create a new WebSocketClient
      *
-     * @param sslContextFactory
-     *            ssl context factory to use
+     * @param bufferPool
+     *            byte buffer pool to use
+     * @deprecated use {@link #WebSocketClient(HttpClient)} instead
      */
-    public WebSocketClient(SslContextFactory sslContextFactory)
+    @Deprecated
+    public WebSocketClient(ByteBufferPool bufferPool)
     {
-        this(sslContextFactory,null);
+        this(null, null, bufferPool);
     }
 
     /**
@@ -166,7 +162,7 @@ public WebSocketClient(SslContextFactory sslContextFactory)
     @Deprecated
     public WebSocketClient(SslContextFactory sslContextFactory, Executor executor)
     {
-        this(sslContextFactory,executor,new MappedByteBufferPool());
+        this(sslContextFactory, executor, null);
     }
 
     /**
@@ -178,7 +174,7 @@ public WebSocketClient(SslContextFactory sslContextFactory, Executor executor)
      */
     public WebSocketClient(WebSocketContainerScope scope)
     {
-        this(scope.getSslContextFactory(),scope.getExecutor(),scope.getBufferPool(),scope.getObjectFactory());
+        this(scope, null, null, null);
     }
 
     /**
@@ -193,7 +189,7 @@ public WebSocketClient(WebSocketContainerScope scope)
      */
     public WebSocketClient(WebSocketContainerScope scope, SslContextFactory sslContextFactory)
     {
-        this(sslContextFactory,scope.getExecutor(),scope.getBufferPool(),scope.getObjectFactory());
+        this(sslContextFactory, scope.getExecutor(), scope.getBufferPool(), scope.getObjectFactory());
     }
 
     /**
@@ -209,7 +205,7 @@ public WebSocketClient(WebSocketContainerScope scope, SslContextFactory sslConte
      */
     public WebSocketClient(SslContextFactory sslContextFactory, Executor executor, ByteBufferPool bufferPool)
     {
-        this(sslContextFactory,executor,bufferPool,new DecoratedObjectFactory());
+        this(sslContextFactory, executor, bufferPool, null);
     }
 
     /**
@@ -227,17 +223,8 @@ public WebSocketClient(SslContextFactory sslContextFactory, Executor executor, B
      */
     private WebSocketClient(SslContextFactory sslContextFactory, Executor executor, ByteBufferPool bufferPool, DecoratedObjectFactory objectFactory)
     {
-        this.httpClient = new HttpClient(sslContextFactory);
-        this.httpClient.setExecutor(executor);
-        this.httpClient.setByteBufferPool(bufferPool);
+        this(new SimpleContainerScope(new WebSocketPolicy(WebSocketBehavior.CLIENT), bufferPool, executor, sslContextFactory, objectFactory));
         addBean(this.httpClient);
-        
-        this.containerScope = new SimpleContainerScope(WebSocketPolicy.newClientPolicy(), bufferPool, objectFactory);
-
-        this.extensionRegistry = new WebSocketExtensionFactory(containerScope);
-
-        this.eventDriverFactory = new EventDriverFactory(containerScope);
-        this.sessionFactory = new WebSocketSessionFactory(containerScope);
     }
 
     /**
@@ -271,20 +258,7 @@ public WebSocketClient(final WebSocketContainerScope scope, EventDriverFactory e
      */
     public WebSocketClient(final WebSocketContainerScope scope, EventDriverFactory eventDriverFactory, SessionFactory sessionFactory, HttpClient httpClient)
     {
-        WebSocketContainerScope clientScope;
-        if (scope.getPolicy().getBehavior() == WebSocketBehavior.CLIENT)
-        {
-            clientScope = scope;
-        }
-        else
-        {
-            // We need to wrap the scope
-            clientScope = new DelegatedContainerScope(WebSocketPolicy.newClientPolicy(), scope);
-        }
-        
-        this.containerScope = clientScope;
-        
-        if(httpClient == null)
+        if (httpClient == null)
         {
             this.httpClient = HttpClientProvider.get(scope);
             addBean(this.httpClient);
@@ -293,11 +267,15 @@ public WebSocketClient(final WebSocketContainerScope scope, EventDriverFactory e
         {
             this.httpClient = httpClient;
         }
-        
-        this.extensionRegistry = new WebSocketExtensionFactory(containerScope);
-        
-        this.eventDriverFactory = eventDriverFactory;
-        this.sessionFactory = sessionFactory;
+
+        // Ensure we get a Client version of the policy.
+        this.policy = scope.getPolicy().clonePolicy(WebSocketBehavior.CLIENT);
+        // Support Late Binding of Object Factory (for CDI)
+        this.objectFactorySupplier = () -> scope.getObjectFactory();
+        this.extensionRegistry = new WebSocketExtensionFactory(this);
+
+        this.eventDriverFactory = eventDriverFactory == null ? new EventDriverFactory(this) : eventDriverFactory;
+        this.sessionFactory = sessionFactory == null ? new WebSocketSessionFactory(this) : sessionFactory;
     }
 
     public Future<Session> connect(Object websocket, URI toUri) throws IOException
@@ -439,7 +417,7 @@ public boolean isDispatchIO()
      */
     public long getAsyncWriteTimeout()
     {
-        return this.containerScope.getPolicy().getAsyncWriteTimeout();
+        return getPolicy().getAsyncWriteTimeout();
     }
 
     public SocketAddress getBindAddress()
@@ -548,7 +526,7 @@ public long getMaxTextMessageSize()
     @Override
     public DecoratedObjectFactory getObjectFactory()
     {
-        return this.containerScope.getObjectFactory();
+        return this.objectFactorySupplier.get();
     }
 
     public Set<WebSocketSession> getOpenSessions()
@@ -559,7 +537,7 @@ public Set<WebSocketSession> getOpenSessions()
     @Override
     public WebSocketPolicy getPolicy()
     {
-        return this.containerScope.getPolicy();
+        return this.policy;
     }
 
     public Scheduler getScheduler()
@@ -753,11 +731,27 @@ public boolean isStopAtShutdown()
         return stopAtShutdown;
     }
 
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o) return true;
+        if (!(o instanceof WebSocketClient)) return false;
+        WebSocketClient that = (WebSocketClient) o;
+        return Objects.equals(this.httpClient, that.httpClient) &&
+                Objects.equals(this.policy, that.policy);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(httpClient, policy);
+    }
+
     @Override
     public String toString()
     {
         final StringBuilder sb = new StringBuilder("WebSocketClient@");
-        sb.append(Integer.toHexString(id));
+        sb.append(Integer.toHexString(hashCode()));
         sb.append("[httpClient=").append(httpClient);
         sb.append(",openSessions.size=");
         sb.append(getOpenSessions().size());