Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merged branch 'jetty-9.3.x' into 'jetty-9.4.x'.

This commit is contained in:
Simone Bordet 2016-06-13 23:39:23 +02:00
parent 831791bdf4 ebee9f12f9
commit 3dca610f7b
6 changed files with 203 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
jetty-proxy/src/test/java/org/eclipse/jetty/proxy/ConnectHandlerTest.java
View File

@ -588,6 +588,7 @@ public class ConnectHandlerTest extends AbstractConnectHandlerTest
"\r\n";
try (Socket socket = newSocket())
{
socket.setSoTimeout((int)idleTimeout);
OutputStream output = socket.getOutputStream();
InputStream input = socket.getInputStream();

35
jetty-websocket/javax-websocket-client-impl/src/main/java/org/eclipse/jetty/websocket/jsr356/ClientContainer.java
View File

@ -20,13 +20,11 @@ package org.eclipse.jetty.websocket.jsr356;
import java.io.IOException;
import java.net.URI;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
@ -74,18 +72,15 @@ import org.eclipse.jetty.websocket.jsr356.metadata.EndpointMetadata;
public class ClientContainer extends ContainerLifeCycle implements WebSocketContainer, WebSocketContainerScope
{
private static final Logger LOG = Log.getLogger(ClientContainer.class);
/** The delegated Container Scope */
private final WebSocketContainerScope scopeDelegate;
/** Tracking all primitive decoders for the container */
private final DecoderFactory decoderFactory;
/** Tracking all primitive encoders for the container */
private final EncoderFactory encoderFactory;
/** Tracking for all declared Client endpoints */
private final Map<Class<?>, EndpointMetadata> endpointClientMetadataCache;
/** Tracking for all open Sessions */
private Set<Session> openSessions = new CopyOnWriteArraySet<>();
/** The jetty websocket client in use for this container */
private WebSocketClient client;
@ -95,11 +90,11 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
this(new SimpleContainerScope(WebSocketPolicy.newClientPolicy()));
client.setDaemon(true);
}
public ClientContainer(WebSocketContainerScope scope)
{
boolean trustAll = Boolean.getBoolean("org.eclipse.jetty.websocket.jsr356.ssl-trust-all");
this.scopeDelegate = scope;
client = new WebSocketClient(scope, new SslContextFactory(trustAll));
client.setEventDriverFactory(new JsrEventDriverFactory(client.getPolicy()));
@ -112,7 +107,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
ShutdownThread.register(this);
}
private Session connect(EndpointInstance instance, URI path) throws IOException
{
Objects.requireNonNull(instance,"EndpointInstance cannot be null");
@ -190,18 +185,18 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
EndpointInstance instance = newClientEndpointInstance(endpoint,null);
return connect(instance,path);
}
@Override
protected void doStart() throws Exception
{
super.doStart();
// Initialize the default decoder / encoder factories
EmptyClientEndpointConfig empty = new EmptyClientEndpointConfig();
this.decoderFactory.init(empty);
this.encoderFactory.init(empty);
}
@Override
protected void doStop() throws Exception
{
@ -223,11 +218,9 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
public EndpointMetadata getClientEndpointMetadata(Class<?> endpoint, EndpointConfig config)
{
EndpointMetadata metadata = null;
synchronized (endpointClientMetadataCache)
{
metadata = endpointClientMetadataCache.get(endpoint);
EndpointMetadata metadata = endpointClientMetadataCache.get(endpoint);
if (metadata != null)
{
@ -257,7 +250,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
err.append(endpoint.getName());
err.append("] does not extend @").append(ClientEndpoint.class.getName());
err.append(" or extend from ").append(Endpoint.class.getName());
throw new InvalidWebSocketException("Unable to identify as valid Endpoint: " + endpoint);
throw new InvalidWebSocketException(err.toString());
}
endpointClientMetadataCache.put(endpoint,metadata);
@ -331,7 +324,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
*/
public Set<Session> getOpenSessions()
{
return Collections.unmodifiableSet(this.openSessions);
return new HashSet<>(getBeans(Session.class));
}
@Override
@ -339,7 +332,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
{
return scopeDelegate.getPolicy();
}
@Override
public SslContextFactory getSslContextFactory()
{
@ -381,7 +374,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
{
if (session instanceof Session)
{
this.openSessions.remove((Session)session);
removeBean(session);
}
else
{
@ -395,7 +388,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
{
if (session instanceof Session)
{
this.openSessions.add((Session)session);
addManaged(session);
}
else
{
@ -403,7 +396,7 @@ public class ClientContainer extends ContainerLifeCycle implements WebSocketCont
Session.class.getName());
}
}
@Override
public void setAsyncSendTimeout(long ms)
{

19
jetty-websocket/javax-websocket-server-impl/src/main/java/org/eclipse/jetty/websocket/jsr356/server/ServerContainer.java
View File

@ -19,11 +19,14 @@
package org.eclipse.jetty.websocket.jsr356.server;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
import javax.websocket.DeploymentException;
import javax.websocket.Endpoint;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import javax.websocket.server.ServerEndpointConfig;
@ -60,7 +63,7 @@ public class ServerContainer extends ClientContainer implements javax.websocket.
this.webSocketServerFactory.addSessionFactory(new JsrSessionFactory(this));
addBean(webSocketServerFactory);
}
public EndpointInstance newClientEndpointInstance(Object endpoint, ServerEndpointConfig config, String path)
{
EndpointMetadata metadata = getClientEndpointMetadata(endpoint.getClass(),config);
@ -124,13 +127,13 @@ public class ServerContainer extends ClientContainer implements javax.websocket.
deferredEndpointConfigs.add(config);
}
}
@Override
protected void doStart() throws Exception
{
// Proceed with Normal Startup
super.doStart();
// Process Deferred Endpoints
if (deferredEndpointClasses != null)
{
@ -140,7 +143,7 @@ public class ServerContainer extends ClientContainer implements javax.websocket.
}
deferredEndpointClasses.clear();
}
if (deferredEndpointConfigs != null)
{
for (ServerEndpointConfig config : deferredEndpointConfigs)
@ -245,14 +248,18 @@ public class ServerContainer extends ClientContainer implements javax.websocket.
@Override
public void onSessionClosed(WebSocketSession session)
{
super.onSessionClosed(session);
webSocketServerFactory.onSessionClosed(session);
}
@Override
public void onSessionOpened(WebSocketSession session)
{
super.onSessionOpened(session);
webSocketServerFactory.onSessionOpened(session);
}
@Override
public Set<Session> getOpenSessions()
{
return new HashSet<>(webSocketServerFactory.getBeans(Session.class));
}
}

226
jetty-websocket/javax-websocket-server-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/server/SessionTrackingTest.java
View File

@ -18,11 +18,8 @@
package org.eclipse.jetty.websocket.jsr356.server;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import java.net.URI;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@ -36,23 +33,129 @@ import javax.websocket.server.ServerEndpoint;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.websocket.common.WebSocketSession;
import org.eclipse.jetty.websocket.jsr356.ClientContainer;
import org.eclipse.jetty.websocket.jsr356.server.deploy.WebSocketServerContainerInitializer;
import org.eclipse.jetty.websocket.server.WebSocketServerFactory;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class SessionTrackingTest
{
public static class ClientSocket extends Endpoint
private Server server;
private ServerContainer serverContainer;
private WebSocketServerFactory wsServerFactory;
private URI serverURI;
@Before
public void startServer() throws Exception
{
public Session session;
public CountDownLatch openLatch = new CountDownLatch(1);
public CountDownLatch closeLatch = new CountDownLatch(1);
QueuedThreadPool serverThreads = new QueuedThreadPool();
serverThreads.setName("server");
server = new Server(serverThreads);
ServerConnector serverConnector = new ServerConnector(server);
serverConnector.setPort(0);
server.addConnector(serverConnector);
ServletContextHandler servletContextHandler = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
servletContextHandler.setContextPath("/");
server.setHandler(servletContextHandler);
serverContainer = WebSocketServerContainerInitializer.configureContext(servletContextHandler);
serverContainer.addEndpoint(EchoSocket.class);
wsServerFactory = serverContainer.getBean(WebSocketServerFactory.class);
server.start();
serverURI = new URI("ws://localhost:" + serverConnector.getLocalPort());
}
@After
public void stopServer() throws Exception
{
if (server != null)
server.stop();
}
@Test
public void testAddRemoveSessions() throws Exception
{
// Create Client
ClientContainer clientContainer = new ClientContainer();
QueuedThreadPool clientThreads = new QueuedThreadPool();
clientThreads.setName("client");
clientContainer.getClient().setExecutor(clientThreads);
try
{
CountDownLatch openedLatch = new CountDownLatch(2);
CountDownLatch closedLatch = new CountDownLatch(2);
wsServerFactory.addSessionListener(new WebSocketSession.Listener()
{
@Override
public void onOpened(WebSocketSession session)
{
openedLatch.countDown();
}
@Override
public void onClosed(WebSocketSession session)
{
closedLatch.countDown();
}
});
clientContainer.start();
// Establish connections
ClientSocket cli1 = new ClientSocket();
clientContainer.connectToServer(cli1, serverURI.resolve("/test"));
cli1.waitForOpen(1, TimeUnit.SECONDS);
// Establish new connection
ClientSocket cli2 = new ClientSocket();
clientContainer.connectToServer(cli2, serverURI.resolve("/test"));
cli2.waitForOpen(1, TimeUnit.SECONDS);
openedLatch.await(5, TimeUnit.SECONDS);
assertServerOpenConnectionCount(2);
// Establish close both connections
cli1.session.close();
cli2.session.close();
cli1.waitForClose(1, TimeUnit.SECONDS);
cli2.waitForClose(1, TimeUnit.SECONDS);
closedLatch.await(5, TimeUnit.SECONDS);
assertServerOpenConnectionCount(0);
}
finally
{
clientContainer.stop();
}
}
private void assertServerOpenConnectionCount(int expectedCount)
{
Set<Session> sessions = serverContainer.getOpenSessions();
int openCount = 0;
for (Session session : sessions)
{
Assert.assertThat("Session.isopen: " + session, session.isOpen(), Matchers.is(true));
openCount++;
}
Assert.assertThat("Open Session Count", openCount, Matchers.is(expectedCount));
}
private static class ClientSocket extends Endpoint
{
private Session session;
private CountDownLatch openLatch = new CountDownLatch(1);
private CountDownLatch closeLatch = new CountDownLatch(1);
@Override
public void onOpen(Session session, EndpointConfig config)
@ -69,12 +172,12 @@ public class SessionTrackingTest
public void waitForOpen(long timeout, TimeUnit unit) throws InterruptedException
{
assertThat("ClientSocket opened",openLatch.await(timeout,unit),is(true));
Assert.assertThat("ClientSocket opened", openLatch.await(timeout, unit), Matchers.is(true));
}
public void waitForClose(long timeout, TimeUnit unit) throws InterruptedException
{
assertThat("ClientSocket opened",closeLatch.await(timeout,unit),is(true));
Assert.assertThat("ClientSocket opened", closeLatch.await(timeout, unit), Matchers.is(true));
}
}
@ -87,99 +190,4 @@ public class SessionTrackingTest
return msg;
}
}
private static Server server;
private static WebSocketServerFactory wsServerFactory;
private static URI serverURI;
@BeforeClass
public static void startServer() throws Exception
{
Server server = new Server();
ServerConnector serverConnector = new ServerConnector(server);
serverConnector.setPort(0);
server.addConnector(serverConnector);
ServletContextHandler servletContextHandler = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
servletContextHandler.setContextPath("/");
server.setHandler(servletContextHandler);
ServerContainer serverContainer = WebSocketServerContainerInitializer.configureContext(servletContextHandler);
serverContainer.addEndpoint(EchoSocket.class);
wsServerFactory = serverContainer.getBean(WebSocketServerFactory.class);
server.start();
String host = serverConnector.getHost();
if (StringUtil.isBlank(host))
{
host = "localhost";
}
serverURI = new URI("ws://" + host + ":" + serverConnector.getLocalPort());
}
@AfterClass
public static void stopServer() throws Exception
{
if (server == null)
{
return;
}
server.stop();
}
@Test
@Ignore
public void testAddRemoveSessions() throws Exception
{
// Create Client
ClientContainer clientContainer = new ClientContainer();
try
{
clientContainer.start();
// Establish connections
ClientSocket cli1 = new ClientSocket();
clientContainer.connectToServer(cli1,serverURI.resolve("/test"));
cli1.waitForOpen(1,TimeUnit.SECONDS);
// Assert open connections
assertServerOpenConnectionCount(1);
// Establish new connection
ClientSocket cli2 = new ClientSocket();
clientContainer.connectToServer(cli2,serverURI.resolve("/test"));
cli2.waitForOpen(1,TimeUnit.SECONDS);
// Assert open connections
assertServerOpenConnectionCount(2);
// Establish close both connections
cli1.session.close();
cli2.session.close();
cli1.waitForClose(1,TimeUnit.SECONDS);
cli2.waitForClose(1,TimeUnit.SECONDS);
// Assert open connections
assertServerOpenConnectionCount(0);
}
finally
{
clientContainer.stop();
}
}
private void assertServerOpenConnectionCount(int expectedCount)
{
Collection<WebSocketSession> sessions = wsServerFactory.getBeans(WebSocketSession.class);
int openCount = 0;
for (WebSocketSession session : sessions)
{
assertThat("Session.isopen: " + session,session.isOpen(),is(true));
openCount++;
}
assertThat("Open Session Count",openCount,is(expectedCount));
}
}

8
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/WebSocketSession.java
View File

@ -474,7 +474,7 @@ public class WebSocketSession extends ContainerLifeCycle implements Session, Rem
{
return new WebSocketRemoteEndpoint(connection,outgoingHandler,getBatchMode());
}
/**
* Open/Activate the session
*/
@ -609,4 +609,10 @@ public class WebSocketSession extends ContainerLifeCycle implements Session, Rem
return builder.toString();
}
public static interface Listener
{
void onOpened(WebSocketSession session);
void onClosed(WebSocketSession session);
}
}

89
jetty-websocket/websocket-server/src/main/java/org/eclipse/jetty/websocket/server/WebSocketServerFactory.java
View File

@ -28,7 +28,9 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
import javax.servlet.ServletContext;
import javax.servlet.ServletException;
@ -88,6 +90,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
* Have the factory maintain 1 and only 1 scheduler. All connections share this scheduler.
*/
private final Scheduler scheduler = new ScheduledExecutorScheduler();
private final List<WebSocketSession.Listener> listeners = new CopyOnWriteArrayList<>();
private final String supportedVersions;
private final WebSocketPolicy defaultPolicy;
private final EventDriverFactory eventDriverFactory;
@ -120,7 +123,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
addBean(scheduler);
addBean(bufferPool);
this.contextClassloader = Thread.currentThread().getContextClassLoader();
this.registeredSocketClasses = new ArrayList<>();
@ -129,7 +132,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
this.eventDriverFactory = new EventDriverFactory(defaultPolicy);
this.bufferPool = bufferPool;
this.extensionFactory = new WebSocketExtensionFactory(this);
this.sessionFactories = new ArrayList<>();
this.sessionFactories.add(new WebSocketSessionFactory(this));
this.creator = this;
@ -153,6 +156,16 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
supportedVersions = rv.toString();
}
public void addSessionListener(WebSocketSession.Listener listener)
{
listeners.add(listener);
}
public void removeSessionListener(WebSocketSession.Listener listener)
{
listeners.remove(listener);
}
@Override
public boolean acceptWebSocket(HttpServletRequest request, HttpServletResponse response) throws IOException
{
@ -166,7 +179,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
try
{
Thread.currentThread().setContextClassLoader(contextClassloader);
// Create Servlet Specific Upgrade Request/Response objects
ServletUpgradeRequest sockreq = new ServletUpgradeRequest(request);
ServletUpgradeResponse sockresp = new ServletUpgradeResponse(response);
@ -188,10 +201,10 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
// Allow Decorators to do their thing
websocketPojo = getObjectFactory().decorate(websocketPojo);
// Get the original HTTPConnection
HttpConnection connection = (HttpConnection)request.getAttribute("org.eclipse.jetty.server.HttpConnection");
// Send the upgrade
EventDriver driver = eventDriverFactory.wrap(websocketPojo);
return upgrade(connection, sockreq, sockresp, driver);
@ -199,13 +212,13 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
catch (URISyntaxException e)
{
throw new IOException("Unable to accept websocket due to mangled URI", e);
}
}
finally
{
Thread.currentThread().setContextClassLoader(old);
}
}
public void addSessionFactory(SessionFactory sessionFactory)
{
if (sessionFactories.contains(sessionFactory))
@ -285,7 +298,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
throw new WebSocketException("Unable to create instance of " + firstClass, e);
}
}
@Override
protected void doStart() throws Exception
{
@ -293,7 +306,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
{
this.objectFactory = new DecoratedObjectFactory();
}
super.doStart();
}
@ -308,7 +321,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
{
return this.creator;
}
@Override
public Executor getExecutor()
{
@ -330,7 +343,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
{
return extensionFactory;
}
public Collection<WebSocketSession> getOpenSessions()
{
return getBeans(WebSocketSession.class);
@ -341,12 +354,12 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
{
return defaultPolicy;
}
@Override
public SslContextFactory getSslContextFactory()
{
/* Not relevant for a Server, as this is defined in the
* Connector configuration
* Connector configuration
*/
return null;
}
@ -358,10 +371,10 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
{
this.objectFactory = new DecoratedObjectFactory();
}
this.executor = context.getServer().getThreadPool();
}
@Override
public void init(ServletContext context) throws ServletException
{
@ -371,7 +384,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
{
this.objectFactory = new DecoratedObjectFactory();
}
// Validate Environment
ContextHandler handler = ContextHandler.getContextHandler(context);
@ -381,7 +394,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
}
this.executor = handler.getServer().getThreadPool();
try
{
// start lifecycle
@ -453,33 +466,33 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
return true;
}
@Override
public void onSessionClosed(WebSocketSession session)
{
removeBean(session);
}
@Override
public void onSessionOpened(WebSocketSession session)
{
addManaged(session);
notifySessionListeners(listener -> listener.onOpened(session));
}
protected String[] parseProtocols(String protocol)
@Override
public void onSessionClosed(WebSocketSession session)
{
if (protocol == null)
removeBean(session);
notifySessionListeners(listener -> listener.onClosed(session));
}
private void notifySessionListeners(Consumer<WebSocketSession.Listener> consumer)
{
for (WebSocketSession.Listener listener : listeners)
{
return new String[]{null};
try
{
consumer.accept(listener);
}
catch (Throwable x)
{
LOG.info("Exception while invoking listener " + listener, x);
}
}
protocol = protocol.trim();
if (protocol.length() == 0)
{
return new String[]{null};
}
String[] passed = protocol.split("\\s*,\\s*");
String[] protocols = new String[passed.length + 1];
System.arraycopy(passed, 0, protocols, 0, passed.length);
return protocols;
}
@Override
@ -594,7 +607,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
WebSocketSession session = createSession(request.getRequestURI(), driver, wsConnection);
session.setPolicy(driver.getPolicy());
session.setUpgradeRequest(request);
// set true negotiated extension list back to response
// set true negotiated extension list back to response
response.setExtensions(extensionStack.getNegotiatedExtensions());
session.setUpgradeResponse(response);
wsConnection.addListener(session);
@ -610,7 +623,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
// Start Components
session.addManaged(extensionStack);
this.addManaged(session);
if (session.isFailed())
{
throw new IOException("Session failed to start");
@ -624,7 +637,7 @@ public class WebSocketServerFactory extends ContainerLifeCycle implements WebSoc
if (getSendServerVersion(connector))
response.setHeader("Server",HttpConfiguration.SERVER_VERSION);
// Process (version specific) handshake response
handshaker.doHandshakeResponse(request, response);