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

Issue #3498 - improve testing for websocket suspend 

Signed-off-by: lachan-roberts <lachlan@webtide.com>
This commit is contained in:
lachan-roberts 2019-05-07 11:00:32 +10:00
parent 6c6646286d
commit 7c7932a475
2 changed files with 174 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

174
jetty-websocket/jetty-websocket-tests/src/test/java/org/eclipse/jetty/websocket/tests/SuspendResumeTest.java Normal file
View File

@ -0,0 +1,174 @@
//
// ========================================================================
// Copyright (c) 1995-2019 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.tests;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.SuspendToken;
import org.eclipse.jetty.websocket.api.annotations.WebSocket;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class SuspendResumeTest
{
@WebSocket
public static class SuspendSocket extends EventSocket
{
volatile SuspendToken suspendToken = null;
@Override
public void onMessage(String message) throws IOException
{
if ("suspend".equals(message))
suspendToken = session.suspend();
super.onMessage(message);
}
@Override
public void onError(Throwable cause)
{
super.onError(cause);
cause.printStackTrace();
}
}
public class UpgradeServlet extends WebSocketServlet
{
@Override
public void configure(WebSocketServletFactory factory)
{
factory.setCreator(((req, resp) -> serverSocket));
}
}
private Server server = new Server();
private WebSocketClient client = new WebSocketClient();
private SuspendSocket serverSocket = new SuspendSocket();
private ServerConnector connector;
@BeforeEach
public void start() throws Exception
{
connector = new ServerConnector(server);
server.addConnector(connector);
ServletContextHandler contextHandler = new ServletContextHandler(ServletContextHandler.SESSIONS);
contextHandler.setContextPath("/");
server.setHandler(contextHandler);
contextHandler.addServlet(new ServletHolder(new UpgradeServlet()), "/suspend");
server.start();
client.start();
}
@AfterEach
public void stop() throws Exception
{
client.stop();
server.stop();
}
@Test
public void testSuspendWhenProcessingFrame() throws Exception
{
URI uri = new URI("ws://localhost:"+connector.getLocalPort()+"/suspend");
EventSocket clientSocket = new EventSocket();
Future<Session> connect = client.connect(clientSocket, uri);
connect.get(5, TimeUnit.SECONDS);
clientSocket.session.getRemote().sendString("suspend");
clientSocket.session.getRemote().sendString("suspend");
clientSocket.session.getRemote().sendString("hello world");
assertThat(serverSocket.receivedMessages.poll(5, TimeUnit.SECONDS), is("suspend"));
assertNull(serverSocket.receivedMessages.poll(1, TimeUnit.SECONDS));
serverSocket.suspendToken.resume();
assertThat(serverSocket.receivedMessages.poll(5, TimeUnit.SECONDS), is("suspend"));
assertNull(serverSocket.receivedMessages.poll(1, TimeUnit.SECONDS));
serverSocket.suspendToken.resume();
assertThat(serverSocket.receivedMessages.poll(5, TimeUnit.SECONDS), is("hello world"));
assertNull(serverSocket.receivedMessages.poll(1, TimeUnit.SECONDS));
// make sure both sides are closed
clientSocket.session.close();
assertTrue(clientSocket.closed.await(5, TimeUnit.SECONDS));
assertTrue(serverSocket.closed.await(5, TimeUnit.SECONDS));
// check no errors occurred
assertNull(clientSocket.error);
assertNull(serverSocket.error);
}
@Test
public void testExternalSuspend() throws Exception
{
URI uri = new URI("ws://localhost:"+connector.getLocalPort()+"/suspend");
EventSocket clientSocket = new EventSocket();
Future<Session> connect = client.connect(clientSocket, uri);
connect.get(5, TimeUnit.SECONDS);
// verify connection by sending a message from server to client
assertTrue(serverSocket.open.await(5, TimeUnit.SECONDS));
serverSocket.session.getRemote().sendString("verification");
assertThat(clientSocket.receivedMessages.poll(5, TimeUnit.SECONDS), is("verification"));
// suspend the client so that no read events occur
SuspendToken suspendToken = clientSocket.session.suspend();
// verify client can still send messages
clientSocket.session.getRemote().sendString("message-from-client");
assertThat(serverSocket.receivedMessages.poll(5, TimeUnit.SECONDS), is("message-from-client"));
// the message is not received as it is suspended
serverSocket.session.getRemote().sendString("message-from-server");
assertNull(clientSocket.receivedMessages.poll(2, TimeUnit.SECONDS));
// client should receive message after it resumes
suspendToken.resume();
assertThat(clientSocket.receivedMessages.poll(5, TimeUnit.SECONDS), is("message-from-server"));
// make sure both sides are closed
clientSocket.session.close();
assertTrue(clientSocket.closed.await(5, TimeUnit.SECONDS));
assertTrue(serverSocket.closed.await(5, TimeUnit.SECONDS));
// check no errors occurred
assertNull(clientSocket.error);
assertNull(serverSocket.error);
}
}

155
jetty-websocket/websocket-server/src/test/java/org/eclipse/jetty/websocket/server/SuspendResumeTest.java
View File

@ -1,155 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2019 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.server;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.fail;
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.SuspendToken;
import org.eclipse.jetty.websocket.api.WriteCallback;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
import org.eclipse.jetty.websocket.api.annotations.WebSocket;
import org.eclipse.jetty.websocket.common.WebSocketFrame;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.test.BlockheadClient;
import org.eclipse.jetty.websocket.common.test.BlockheadClientRequest;
import org.eclipse.jetty.websocket.common.test.BlockheadConnection;
import org.eclipse.jetty.websocket.common.test.Timeouts;
import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
public class SuspendResumeTest
{
@WebSocket
public static class EchoSocket
{
private Session session;
@OnWebSocketConnect
public void onConnect(Session session)
{
this.session = session;
}
@OnWebSocketMessage
public void onMessage(String message)
{
SuspendToken suspendToken = this.session.suspend();
this.session.getRemote().sendString(message,
new WriteCallback()
{
@Override
public void writeSuccess()
{
suspendToken.resume();
}
@Override
public void writeFailed(Throwable t)
{
fail(t);
}
});
}
}
public static class EchoCreator implements WebSocketCreator
{
@Override
public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp)
{
return new EchoSocket();
}
}
public static class EchoServlet extends WebSocketServlet
{
private static final long serialVersionUID = 1L;
@Override
public void configure(WebSocketServletFactory factory)
{
factory.setCreator(new EchoCreator());
}
}
private static SimpleServletServer server;
private static BlockheadClient client;
@BeforeAll
public static void startServer() throws Exception
{
server = new SimpleServletServer(new EchoServlet());
server.start();
}
@BeforeAll
public static void startClient() throws Exception
{
client = new BlockheadClient();
client.setIdleTimeout(TimeUnit.SECONDS.toMillis(2));
client.start();
}
@AfterAll
public static void stopClient() throws Exception
{
client.stop();
}
@AfterAll
public static void stopServer()
{
server.stop();
}
@Test
public void testSuspendResume() throws Exception
{
BlockheadClientRequest request = client.newWsRequest(server.getServerUri());
Future<BlockheadConnection> connFut = request.sendAsync();
try (BlockheadConnection clientConn = connFut.get(Timeouts.CONNECT, Timeouts.CONNECT_UNIT))
{
clientConn.write(new TextFrame().setPayload("echo1"));
clientConn.write(new TextFrame().setPayload("echo2"));
LinkedBlockingQueue<WebSocketFrame> frames = clientConn.getFrameQueue();
WebSocketFrame tf = frames.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
assertThat(EchoSocket.class.getSimpleName() + ".onMessage()", tf.getPayloadAsUTF8(), is("echo1"));
tf = frames.poll(Timeouts.POLL_EVENT, Timeouts.POLL_EVENT_UNIT);
assertThat(EchoSocket.class.getSimpleName() + ".onMessage()", tf.getPayloadAsUTF8(), is("echo2"));
}
}
}