A {@link HttpClientTransport} that can dynamically switch among different application protocols.
@@ -79,6 +81,8 @@ */ public class HttpClientTransportDynamic extends AbstractConnectorHttpClientTransport { + private static final Logger LOG = LoggerFactory.getLogger(HttpClientTransportDynamic.class); + private final ListThe client-side component that connects to server sockets.
+ *ClientConnector delegates the handling of {@link SocketChannel}s + * to a {@link SelectorManager}, and centralizes the configuration of + * necessary components such as the executor, the scheduler, etc.
+ *ClientConnector offers a low-level API that can be used to + * connect {@link SocketChannel}s to listening servers via the + * {@link #connect(SocketAddress, Map)} method.
+ *However, a ClientConnector instance is typically just configured + * and then passed to an HttpClient transport, so that applications + * can use high-level APIs to make HTTP requests to servers:
+ *+ * // Create a ClientConnector instance. + * ClientConnector connector = new ClientConnector(); + * + * // Configure the ClientConnector. + * connector.setSelectors(1); + * connector.setSslContextFactory(new SslContextFactory.Client()); + * + * // Pass it to the HttpClient transport. + * HttpClientTransport transport = new HttpClientTransportDynamic(connector); + * HttpClient httpClient = new HttpClient(transport); + * httpClient.start(); + *+ */ +@ManagedObject public class ClientConnector extends ContainerLifeCycle { public static final String CLIENT_CONNECTOR_CONTEXT_KEY = "org.eclipse.jetty.client.connector"; @@ -43,6 +78,18 @@ public class ClientConnector extends ContainerLifeCycle public static final String CONNECTION_PROMISE_CONTEXT_KEY = CLIENT_CONNECTOR_CONTEXT_KEY + ".connectionPromise"; private static final Logger LOG = LoggerFactory.getLogger(ClientConnector.class); + /** + *
Creates a ClientConnector configured to connect via Unix-Domain sockets to the given Unix-Domain path
+ * + * @param path the Unix-Domain path to connect to + * @return a ClientConnector that connects to the given Unix-Domain path + */ + public static ClientConnector forUnixDomain(Path path) + { + return new ClientConnector(SocketChannelWithAddress.Factory.forUnixDomain(path)); + } + + private final SocketChannelWithAddress.Factory factory; private Executor executor; private Scheduler scheduler; private ByteBufferPool byteBufferPool; @@ -53,7 +100,21 @@ public class ClientConnector extends ContainerLifeCycle private Duration connectTimeout = Duration.ofSeconds(5); private Duration idleTimeout = Duration.ofSeconds(30); private SocketAddress bindAddress; + private boolean tcpNoDelay = true; private boolean reuseAddress = true; + private boolean reusePort; + private int receiveBufferSize = -1; + private int sendBufferSize = -1; + + public ClientConnector() + { + this((address, context) -> new SocketChannelWithAddress(SocketChannel.open(), address)); + } + + private ClientConnector(SocketChannelWithAddress.Factory factory) + { + this.factory = Objects.requireNonNull(factory); + } public Executor getExecutor() { @@ -107,6 +168,10 @@ public void setSslContextFactory(SslContextFactory.Client sslContextFactory) this.sslContextFactory = sslContextFactory; } + /** + * @return the number of NIO selectors + */ + @ManagedAttribute("The number of NIO selectors") public int getSelectors() { return selectors; @@ -119,6 +184,10 @@ public void setSelectors(int selectors) this.selectors = selectors; } + /** + * @return whether {@link #connect(SocketAddress, Map)} operations are performed in blocking mode + */ + @ManagedAttribute("Whether connect operations are performed in blocking mode") public boolean isConnectBlocking() { return connectBlocking; @@ -129,6 +198,10 @@ public void setConnectBlocking(boolean connectBlocking) this.connectBlocking = connectBlocking; } + /** + * @return the timeout of {@link #connect(SocketAddress, Map)} operations + */ + @ManagedAttribute("The timeout of connect operations") public Duration getConnectTimeout() { return connectTimeout; @@ -141,6 +214,10 @@ public void setConnectTimeout(Duration connectTimeout) selectorManager.setConnectTimeout(connectTimeout.toMillis()); } + /** + * @return the max duration for which a connection can be idle (that is, without traffic of bytes in either direction) + */ + @ManagedAttribute("The duration for which a connection can be idle") public Duration getIdleTimeout() { return idleTimeout; @@ -151,26 +228,120 @@ public void setIdleTimeout(Duration idleTimeout) this.idleTimeout = idleTimeout; } + /** + * @return the address to bind a socket to before the connect operation + */ + @ManagedAttribute("The socket address to bind sockets to before the connect operation") public SocketAddress getBindAddress() { return bindAddress; } + /** + *Sets the bind address of sockets before the connect operation.
+ *In multi-homed hosts, you may want to connect from a specific address:
+ *
+ * clientConnector.setBindAddress(new InetSocketAddress("127.0.0.2", 0));
+ *
+ * Note the use of the port {@code 0} to indicate that a different ephemeral port + * should be used for each different connection.
+ *In the rare cases where you want to use the same port for all connections, + * you must also call {@link #setReusePort(boolean) setReusePort(true)}.
+ * + * @param bindAddress the socket address to bind to before the connect operation + */ public void setBindAddress(SocketAddress bindAddress) { this.bindAddress = bindAddress; } + /** + * @return whether small TCP packets are sent without delay + */ + @ManagedAttribute("Whether small TCP packets are sent without delay") + public boolean isTCPNoDelay() + { + return tcpNoDelay; + } + + public void setTCPNoDelay(boolean tcpNoDelay) + { + this.tcpNoDelay = tcpNoDelay; + } + + /** + * @return whether rebinding is allowed with sockets in tear-down states + */ + @ManagedAttribute("Whether rebinding is allowed with sockets in tear-down states") public boolean getReuseAddress() { return reuseAddress; } + /** + *Sets whether it is allowed to bind a socket to a socket address + * that may be in use by another socket in tear-down state, for example + * in TIME_WAIT state.
+ *This is useful when ClientConnector is restarted: an existing connection + * may still be using a network address (same host and same port) that is also + * chosen for a new connection.
+ * + * @param reuseAddress whether rebinding is allowed with sockets in tear-down states + * @see #setReusePort(boolean) + */ public void setReuseAddress(boolean reuseAddress) { this.reuseAddress = reuseAddress; } + /** + * @return whether binding to same host and port is allowed + */ + @ManagedAttribute("Whether binding to same host and port is allowed") + public boolean isReusePort() + { + return reusePort; + } + + /** + *Sets whether it is allowed to bind multiple sockets to the same + * socket address (same host and same port).
+ * + * @param reusePort whether binding to same host and port is allowed + */ + public void setReusePort(boolean reusePort) + { + this.reusePort = reusePort; + } + + /** + * @return the receive buffer size in bytes, or -1 for the default value + */ + @ManagedAttribute("The receive buffer size in bytes") + public int getReceiveBufferSize() + { + return receiveBufferSize; + } + + public void setReceiveBufferSize(int receiveBufferSize) + { + this.receiveBufferSize = receiveBufferSize; + } + + /** + * @return the send buffer size in bytes, or -1 for the default value + */ + @ManagedAttribute("The send buffer size in bytes") + public int getSendBufferSize() + { + return sendBufferSize; + } + + public void setSendBufferSize(int sendBufferSize) + { + this.sendBufferSize = sendBufferSize; + } + @Override protected void doStart() throws Exception { @@ -221,20 +392,18 @@ public void connect(SocketAddress address, MapA pair/record holding a {@link SocketChannel} and a {@link SocketAddress} to connect to.
+ */ + private static class SocketChannelWithAddress + { + private final SocketChannel channel; + private final SocketAddress address; + + private SocketChannelWithAddress(SocketChannel channel, SocketAddress address) + { + this.channel = channel; + this.address = address; + } + + private SocketChannel getSocketChannel() + { + return channel; + } + + private SocketAddress getSocketAddress() + { + return address; + } + + /** + *A factory for {@link SocketChannelWithAddress} instances.
+ */ + private interface Factory + { + private static Factory forUnixDomain(Path path) + { + return (address, context) -> + { + try + { + ProtocolFamily family = Enum.valueOf(StandardProtocolFamily.class, "UNIX"); + SocketChannel socketChannel = (SocketChannel)SocketChannel.class.getMethod("open", ProtocolFamily.class).invoke(null, family); + Class addressClass = Class.forName("java.net.UnixDomainSocketAddress"); + SocketAddress socketAddress = (SocketAddress)addressClass.getMethod("of", Path.class).invoke(null, path); + return new SocketChannelWithAddress(socketChannel, socketAddress); + } + catch (Throwable x) + { + String message = "Unix-Domain SocketChannels are available starting from Java 16, your Java version is: " + JavaVersion.VERSION; + throw new UnsupportedOperationException(message, x); + } + }; + } + + /** + *Creates a new {@link SocketChannel} to connect to a {@link SocketAddress} + * derived from the input socket address.
+ *The input socket address represents the destination socket address to + * connect to, as it is typically specified by a URI authority, for example + * {@code localhost:8080} if the URI is {@code http://localhost:8080/path}.
+ *However, the returned socket address may be different as the implementation + * may use a Unix-Domain socket address to physically connect to the virtual + * destination socket address given as input.
+ *The return type is a pair/record holding the socket channel and the + * socket address, with the socket channel not yet connected. + * The implementation of this methods must not call + * {@link SocketChannel#connect(SocketAddress)}, as this is done later, + * after configuring the socket, by the {@link ClientConnector} implementation.
+ * + * @param address the destination socket address, typically specified in a URI + * @param context the context to create the new socket channel + * @return a new {@link SocketChannel} with an associated {@link SocketAddress} to connect to + * @throws IOException if the socket channel or the socket address cannot be created + */ + public SocketChannelWithAddress newSocketChannelWithAddress(SocketAddress address, MapThreads are reserved lazily, with a new reserved thread being allocated from a - * wrapped {@link Executor} when an execution fails. If the {@link #setIdleTimeout(long, TimeUnit)} - * is set to non zero (default 1 minute), then the reserved thread pool will shrink by 1 thread - * whenever it has been idle for that period. + *
Threads are reserved lazily, with a new reserved threads being allocated from the
+ * {@link Executor} passed to the constructor. Whenever 1 or more reserved threads have been
+ * idle for more than {@link #getIdleTimeoutMs()} then one reserved thread will return to
+ * the executor.
*/
@ManagedObject("A pool for reserved threads")
-public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExecutor
+public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExecutor, Dumpable
{
private static final Logger LOG = LoggerFactory.getLogger(ReservedThreadExecutor.class);
+ private static final long DEFAULT_IDLE_TIMEOUT = TimeUnit.MINUTES.toNanos(1);
private static final Runnable STOP = new Runnable()
{
@Override
@@ -57,13 +68,13 @@ public String toString()
private final Executor _executor;
private final int _capacity;
- private final ConcurrentLinkedDeque
This configuration configures the WebAppContext server/system classes to * be able to see the org.eclipse.jetty.jaas package. * This class is defined in the webapp package, as it implements the {@link Configuration} interface, - * which is unknown to the jaas package. However, the corresponding {@link ServiceLoader} - * resource is defined in the jaas package, so that this configuration only be - * loaded if the jetty-jaas jars are on the classpath. + * which is unknown to the jaas package. *
*/ public class JaasConfiguration extends AbstractConfiguration diff --git a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JettyWebXmlConfiguration.java b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JettyWebXmlConfiguration.java index aba74957f58b..7dccaaea820a 100644 --- a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JettyWebXmlConfiguration.java +++ b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JettyWebXmlConfiguration.java @@ -24,7 +24,7 @@ /** * JettyWebConfiguration. * - * Looks for XmlConfiguration files in WEB-INF. Searches in order for the first of jetty6-web.xml, jetty-web.xml or web-jetty.xml + * Looks for XmlConfiguration files in WEB-INF. Searches in order for the first of jetty8-web.xml, jetty-web.xml or web-jetty.xml */ public class JettyWebXmlConfiguration extends AbstractConfiguration { diff --git a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JmxConfiguration.java b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JmxConfiguration.java index 2eebbbebad68..ed9cef8fd595 100644 --- a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JmxConfiguration.java +++ b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JmxConfiguration.java @@ -13,8 +13,6 @@ package org.eclipse.jetty.webapp; -import java.util.ServiceLoader; - import org.eclipse.jetty.util.Loader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -24,9 +22,7 @@ *This configuration configures the WebAppContext server/system classes to * be able to see the org.eclipse.jetty.jmx package. This class is defined * in the webapp package, as it implements the {@link Configuration} interface, - * which is unknown to the jmx package. However, the corresponding {@link ServiceLoader} - * resource is defined in the jmx package, so that this configuration only be - * loaded if the jetty-jmx jars are on the classpath. + * which is unknown to the jmx package. *
*/ public class JmxConfiguration extends AbstractConfiguration diff --git a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JndiConfiguration.java b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JndiConfiguration.java index 5c446b1b2097..2de34e0c6fff 100644 --- a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JndiConfiguration.java +++ b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JndiConfiguration.java @@ -13,8 +13,6 @@ package org.eclipse.jetty.webapp; -import java.util.ServiceLoader; - import org.eclipse.jetty.util.Loader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -24,9 +22,7 @@ *This configuration configures the WebAppContext system/server classes to * be able to see the org.eclipse.jetty.jaas package. * This class is defined in the webapp package, as it implements the {@link Configuration} interface, - * which is unknown to the jndi package. However, the corresponding {@link ServiceLoader} - * resource is defined in the jndi package, so that this configuration only be - * loaded if the jetty-jndi jars are on the classpath. + * which is unknown to the jndi package. *
*/ public class JndiConfiguration extends AbstractConfiguration diff --git a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JspConfiguration.java b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JspConfiguration.java index 12333f5bbd1e..dd6d1e3cb6cc 100644 --- a/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JspConfiguration.java +++ b/jetty-webapp/src/main/java/org/eclipse/jetty/webapp/JspConfiguration.java @@ -13,8 +13,6 @@ package org.eclipse.jetty.webapp; -import java.util.ServiceLoader; - import org.eclipse.jetty.util.Loader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -24,9 +22,7 @@ *This configuration configures the WebAppContext server/system classes to * be able to see the org.eclipse.jetty.jsp and org.eclipse.jetty.apache packages. * This class is defined in the webapp package, as it implements the {@link Configuration} interface, - * which is unknown to the jsp package. However, the corresponding {@link ServiceLoader} - * resource is defined in the jsp package, so that this configuration only be - * loaded if the jetty-jsp jars are on the classpath. + * which is unknown to the jsp package. *
*/ public class JspConfiguration extends AbstractConfiguration diff --git a/jetty-webapp/src/test/java/org/eclipse/jetty/webapp/TempDirTest.java b/jetty-webapp/src/test/java/org/eclipse/jetty/webapp/TempDirTest.java index 97803758d4a3..4f95f09fd1d9 100644 --- a/jetty-webapp/src/test/java/org/eclipse/jetty/webapp/TempDirTest.java +++ b/jetty-webapp/src/test/java/org/eclipse/jetty/webapp/TempDirTest.java @@ -27,6 +27,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfSystemProperty; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -289,9 +290,11 @@ public void jettyBaseWorkExists() throws Exception /** * ServletContext.TEMPDIR has invalidString directory value (wrong permission to write into it)
- * IllegalStateException
+ *
+ * Note that if run in the CI environment, the test will fail, because it runs as root,
+ * so we _will_ have permission to write to this directory.
*/
- @Disabled("Jenkins will run as root so we do have permission to write to this directory.")
+ @DisabledIfSystemProperty(named = "env", matches = "ci")
@Test
public void attributeWithInvalidPermissions()
{
diff --git a/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/WebSocketCoreClient.java b/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/WebSocketCoreClient.java
index c8db7f1d82f3..950231f0fa17 100644
--- a/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/WebSocketCoreClient.java
+++ b/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/WebSocketCoreClient.java
@@ -40,13 +40,6 @@ public class WebSocketCoreClient extends ContainerLifeCycle
private final WebSocketComponents components;
private ClassLoader classLoader;
- // TODO: Things to consider for inclusion in this class (or removal if they can be set elsewhere, like HttpClient)
- // - AsyncWrite Idle Timeout
- // - Bind Address
- // - SslContextFactory setup
- // - Connect Timeout
- // - Cookie Store
-
public WebSocketCoreClient()
{
this(null, new WebSocketComponents());
@@ -61,6 +54,8 @@ public WebSocketCoreClient(HttpClient httpClient, WebSocketComponents webSocketC
{
if (httpClient == null)
httpClient = Objects.requireNonNull(HttpClientProvider.get());
+ if (httpClient.getExecutor() == null)
+ httpClient.setExecutor(webSocketComponents.getExecutor());
this.classLoader = Thread.currentThread().getContextClassLoader();
this.httpClient = httpClient;
diff --git a/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/internal/HttpClientProvider.java b/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/internal/HttpClientProvider.java
index f6b855e8525f..d1ea884064ed 100644
--- a/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/internal/HttpClientProvider.java
+++ b/jetty-websocket/websocket-core-client/src/main/java/org/eclipse/jetty/websocket/core/client/internal/HttpClientProvider.java
@@ -14,7 +14,6 @@
package org.eclipse.jetty.websocket.core.client.internal;
import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.util.thread.QueuedThreadPool;
public interface HttpClientProvider
{
@@ -30,11 +29,7 @@ static HttpClient get()
private static HttpClient newDefaultHttpClient()
{
- HttpClient client = new HttpClient();
- QueuedThreadPool threadPool = new QueuedThreadPool();
- threadPool.setName("WebSocketClient@" + client.hashCode());
- client.setExecutor(threadPool);
- return client;
+ return new HttpClient();
}
default HttpClient newHttpClient()
diff --git a/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/CoreSession.java b/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/CoreSession.java
index 7c6bd9bb4ec0..bbaf05c4e5a7 100644
--- a/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/CoreSession.java
+++ b/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/CoreSession.java
@@ -256,7 +256,7 @@ public SocketAddress getRemoteAddress()
@Override
public boolean isOutputOpen()
{
- return false;
+ return true;
}
@Override
diff --git a/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketComponents.java b/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketComponents.java
index f9c6e86e742e..22353c83b825 100644
--- a/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketComponents.java
+++ b/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketComponents.java
@@ -13,6 +13,7 @@
package org.eclipse.jetty.websocket.core;
+import java.util.concurrent.Executor;
import java.util.zip.Deflater;
import org.eclipse.jetty.io.ByteBufferPool;
@@ -22,6 +23,7 @@
import org.eclipse.jetty.util.compression.CompressionPool;
import org.eclipse.jetty.util.compression.DeflaterPool;
import org.eclipse.jetty.util.compression.InflaterPool;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
/**
* A collection of components which are the resources needed for websockets such as
@@ -29,11 +31,12 @@
*/
public class WebSocketComponents extends ContainerLifeCycle
{
- private final DecoratedObjectFactory objectFactory;
- private final WebSocketExtensionRegistry extensionRegistry;
- private final ByteBufferPool bufferPool;
- private final InflaterPool inflaterPool;
- private final DeflaterPool deflaterPool;
+ private final DecoratedObjectFactory _objectFactory;
+ private final WebSocketExtensionRegistry _extensionRegistry;
+ private final Executor _executor;
+ private final ByteBufferPool _bufferPool;
+ private final InflaterPool _inflaterPool;
+ private final DeflaterPool _deflaterPool;
public WebSocketComponents()
{
@@ -43,41 +46,64 @@ public WebSocketComponents()
public WebSocketComponents(WebSocketExtensionRegistry extensionRegistry, DecoratedObjectFactory objectFactory,
ByteBufferPool bufferPool, InflaterPool inflaterPool, DeflaterPool deflaterPool)
{
- this.extensionRegistry = (extensionRegistry == null) ? new WebSocketExtensionRegistry() : extensionRegistry;
- this.objectFactory = (objectFactory == null) ? new DecoratedObjectFactory() : objectFactory;
- this.bufferPool = (bufferPool == null) ? new MappedByteBufferPool() : bufferPool;
- this.inflaterPool = (inflaterPool == null) ? new InflaterPool(CompressionPool.DEFAULT_CAPACITY, true) : inflaterPool;
- this.deflaterPool = (deflaterPool == null) ? new DeflaterPool(CompressionPool.DEFAULT_CAPACITY, Deflater.DEFAULT_COMPRESSION, true) : deflaterPool;
+ this (extensionRegistry, objectFactory, bufferPool, inflaterPool, deflaterPool, null);
+ }
+
+ public WebSocketComponents(WebSocketExtensionRegistry extensionRegistry, DecoratedObjectFactory objectFactory,
+ ByteBufferPool bufferPool, InflaterPool inflaterPool, DeflaterPool deflaterPool, Executor executor)
+ {
+ _extensionRegistry = (extensionRegistry == null) ? new WebSocketExtensionRegistry() : extensionRegistry;
+ _objectFactory = (objectFactory == null) ? new DecoratedObjectFactory() : objectFactory;
+ _bufferPool = (bufferPool == null) ? new MappedByteBufferPool() : bufferPool;
+ _inflaterPool = (inflaterPool == null) ? new InflaterPool(CompressionPool.DEFAULT_CAPACITY, true) : inflaterPool;
+ _deflaterPool = (deflaterPool == null) ? new DeflaterPool(CompressionPool.DEFAULT_CAPACITY, Deflater.DEFAULT_COMPRESSION, true) : deflaterPool;
- addBean(inflaterPool);
- addBean(deflaterPool);
- addBean(bufferPool);
- addBean(extensionRegistry);
- addBean(objectFactory);
+ if (executor == null)
+ {
+ QueuedThreadPool threadPool = new QueuedThreadPool();
+ threadPool.setName("WebSocket@" + hashCode());
+ _executor = threadPool;
+ }
+ else
+ {
+ _executor = executor;
+ }
+
+ addBean(_inflaterPool);
+ addBean(_deflaterPool);
+ addBean(_bufferPool);
+ addBean(_extensionRegistry);
+ addBean(_objectFactory);
+ addBean(_executor);
}
public ByteBufferPool getBufferPool()
{
- return bufferPool;
+ return _bufferPool;
+ }
+
+ public Executor getExecutor()
+ {
+ return _executor;
}
public WebSocketExtensionRegistry getExtensionRegistry()
{
- return extensionRegistry;
+ return _extensionRegistry;
}
public DecoratedObjectFactory getObjectFactory()
{
- return objectFactory;
+ return _objectFactory;
}
public InflaterPool getInflaterPool()
{
- return inflaterPool;
+ return _inflaterPool;
}
public DeflaterPool getDeflaterPool()
{
- return deflaterPool;
+ return _deflaterPool;
}
}
diff --git a/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/messages/DispatchedMessageSink.java b/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/messages/DispatchedMessageSink.java
index 875fab0096bd..57d31e0d3f67 100644
--- a/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/messages/DispatchedMessageSink.java
+++ b/jetty-websocket/websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/messages/DispatchedMessageSink.java
@@ -16,6 +16,7 @@
import java.io.Closeable;
import java.lang.invoke.MethodHandle;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.IO;
@@ -95,10 +96,12 @@ public abstract class DispatchedMessageSink extends AbstractMessageSink
{
private CompletableFuture