Paramaterize WebSocket tests by client and server

Issue: SPR-14527

spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/AbstractWebSocketIntegrationTests.java

@@ -27,7 +27,10 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameter;
 import org.junit.runners.Parameterized.Parameters;
+import reactor.core.publisher.Flux;
+import reactor.util.function.Tuple3;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -40,6 +43,12 @@ import org.springframework.http.server.reactive.bootstrap.TomcatHttpServer;
 import org.springframework.http.server.reactive.bootstrap.UndertowHttpServer;
 import org.springframework.util.SocketUtils;
 import org.springframework.web.reactive.DispatcherHandler;
+import org.springframework.web.reactive.socket.client.JettyWebSocketClient;
+import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
+import org.springframework.web.reactive.socket.client.RxNettyWebSocketClient;
+import org.springframework.web.reactive.socket.client.StandardWebSocketClient;
+import org.springframework.web.reactive.socket.client.UndertowWebSocketClient;
+import org.springframework.web.reactive.socket.client.WebSocketClient;
 import org.springframework.web.reactive.socket.server.support.HandshakeWebSocketService;
 import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
 import org.springframework.web.reactive.socket.server.upgrade.JettyRequestUpgradeStrategy;
@@ -48,6 +57,8 @@ import org.springframework.web.reactive.socket.server.upgrade.RxNettyRequestUpgr
 import org.springframework.web.reactive.socket.server.upgrade.TomcatRequestUpgradeStrategy;
 import org.springframework.web.reactive.socket.server.upgrade.UndertowRequestUpgradeStrategy;
 
+import static org.junit.Assume.assumeFalse;
+
 /**
  * Base class for WebSocket integration tests.
  * Sub-classes must implement {@link #getWebConfigClass()} to return Spring
@@ -62,37 +73,76 @@ public abstract class AbstractWebSocketIntegrationTests {
 	protected int port;
 
 	@Parameter(0)
-	public HttpServer server;
+	public WebSocketClient client;
 
 	@Parameter(1)
-	public Class<?> handlerAdapterConfigClass;
+	public HttpServer server;
 
+	@Parameter(2)
+	public Class<?> serverConfigClass;
 
-	@Parameters(name = "server [{0}]")
+
+	@Parameters(name = "client[{0}] - server [{1}]")
 	public static Object[][] arguments() {
+
 		File base = new File(System.getProperty("java.io.tmpdir"));
-		return new Object[][] {
-				{new ReactorHttpServer(), ReactorNettyConfig.class},
-				{new RxNettyHttpServer(), RxNettyConfig.class},
-				{new TomcatHttpServer(base.getAbsolutePath(), WsContextListener.class), TomcatConfig.class},
-				{new UndertowHttpServer(), UndertowConfig.class},
-				{new JettyHttpServer(), JettyConfig.class}
-		};
+
+		Flux<? extends WebSocketClient> clients = Flux.concat(
+				Flux.just(new StandardWebSocketClient()).repeat(5),
+				Flux.just(new JettyWebSocketClient()).repeat(5),
+				Flux.just(new ReactorNettyWebSocketClient()).repeat(5),
+				Flux.just(new RxNettyWebSocketClient()).repeat(5),
+				Flux.just(new UndertowWebSocketClient()).repeat(5));
+
+		Flux<? extends HttpServer> servers = Flux.just(
+				new TomcatHttpServer(base.getAbsolutePath(), WsContextListener.class),
+				new JettyHttpServer(),
+				new ReactorHttpServer(),
+				new RxNettyHttpServer(),
+				new UndertowHttpServer()).repeat(5);
+
+		Flux<? extends Class<?>> configs = Flux.just(
+				TomcatConfig.class,
+				JettyConfig.class,
+				ReactorNettyConfig.class,
+				RxNettyConfig.class,
+				UndertowConfig.class).repeat(5);
+
+		return Flux.zip(clients, servers, configs)
+				.map(Tuple3::toArray)
+				.collectList()
+				.block()
+				.toArray(new Object[25][2]);
 	}
 
 
 	@Before
 	public void setup() throws Exception {
+
+		// TODO
+		// Caused by: java.io.IOException: Upgrade responses cannot have a transfer coding
+		// at org.xnio.http.HttpUpgrade$HttpUpgradeState.handleUpgrade(HttpUpgrade.java:490)
+		// at org.xnio.http.HttpUpgrade$HttpUpgradeState.access$1200(HttpUpgrade.java:165)
+		// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:461)
+		// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:400)
+		// at org.xnio.ChannelListeners.invokeChannelListener(ChannelListeners.java:92)
+
+		assumeFalse(this.client instanceof UndertowWebSocketClient && this.server instanceof RxNettyHttpServer);
+
 		this.port = SocketUtils.findAvailableTcpPort();
 		this.server.setPort(this.port);
 		this.server.setHandler(createHttpHandler());
 		this.server.afterPropertiesSet();
 		this.server.start();
+
+		if (this.client instanceof Lifecycle) {
+			((Lifecycle) this.client).start();
+		}
 	}
 
 	private HttpHandler createHttpHandler() {
 		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
-		context.register(DispatcherConfig.class, this.handlerAdapterConfigClass);
+		context.register(DispatcherConfig.class, this.serverConfigClass);
 		context.register(getWebConfigClass());
 		context.refresh();
 		return DispatcherHandler.toHttpHandler(context);
@@ -102,6 +152,9 @@ public abstract class AbstractWebSocketIntegrationTests {
 
 	@After
 	public void tearDown() throws Exception {
+		if (this.client instanceof Lifecycle) {
+			((Lifecycle) this.client).stop();
+		}
 		this.server.stop();
 	}
 

spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java

@@ -15,13 +15,11 @@
  */
 package org.springframework.web.reactive.socket.server;
 
-import java.net.URISyntaxException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.hamcrest.Matchers;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -32,7 +30,6 @@ import reactor.core.publisher.ReplayProcessor;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.HttpHeaders;
-import org.springframework.http.server.reactive.bootstrap.RxNettyHttpServer;
 import org.springframework.web.reactive.HandlerMapping;
 import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
 import org.springframework.web.reactive.socket.HandshakeInfo;
@@ -41,13 +38,10 @@ import org.springframework.web.reactive.socket.WebSocketMessage;
 import org.springframework.web.reactive.socket.WebSocketSession;
 import org.springframework.web.reactive.socket.client.JettyWebSocketClient;
 import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
-import org.springframework.web.reactive.socket.client.RxNettyWebSocketClient;
-import org.springframework.web.reactive.socket.client.StandardWebSocketClient;
-import org.springframework.web.reactive.socket.client.UndertowWebSocketClient;
-import org.springframework.web.reactive.socket.client.WebSocketClient;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assume.assumeFalse;
 
 /**
  * Integration tests with server-side {@link WebSocketHandler}s.
@@ -64,44 +58,7 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
 
 
 	@Test
-	public void echoReactorClient() throws Exception {
-		testEcho(new ReactorNettyWebSocketClient());
-	}
-
-	@Test
-	public void echoRxNettyClient() throws Exception {
-		testEcho(new RxNettyWebSocketClient());
-	}
-
-	@Test
-	public void echoJettyClient() throws Exception {
-		JettyWebSocketClient client = new JettyWebSocketClient();
-		client.start();
-		testEcho(client);
-		client.stop();
-	}
-
-	@Test
-	public void echoStandardClient() throws Exception {
-		testEcho(new StandardWebSocketClient());
-	}
-
-	@Test
-	public void echoUndertowClient() throws Exception {
-		if (server instanceof RxNettyHttpServer) {
-			// Caused by: java.io.IOException: Upgrade responses cannot have a transfer coding
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState.handleUpgrade(HttpUpgrade.java:490)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState.access$1200(HttpUpgrade.java:165)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:461)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:400)
-			// at org.xnio.ChannelListeners.invokeChannelListener(ChannelListeners.java:92)
-
-			return;
-		}
-		testEcho(new UndertowWebSocketClient());
-	}
-
-	private void testEcho(WebSocketClient client) throws URISyntaxException {
+	public void echo() throws Exception {
 		int count = 100;
 		Flux<String> input = Flux.range(1, count).map(index -> "msg-" + index);
 		ReplayProcessor<Object> output = ReplayProcessor.create(count);
@@ -118,45 +75,12 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
 	}
 
 	@Test
-	@Ignore("https://github.com/reactor/reactor-netty/issues/20")
-	public void subProtocolReactorClient() throws Exception {
-		testSubProtocol(new ReactorNettyWebSocketClient());
-	}
-
-	@Test
-	public void subProtocolRxNettyClient() throws Exception {
-		testSubProtocol(new RxNettyWebSocketClient());
-	}
-
-	@Test
-	public void subProtocolJettyClient() throws Exception {
-		JettyWebSocketClient client = new JettyWebSocketClient();
-		client.start();
-		testSubProtocol(client);
-		client.stop();
-	}
+	public void subProtocol() throws Exception {
 
-	@Test
-	public void subProtocolStandardClient() throws Exception {
-		testSubProtocol(new StandardWebSocketClient());
-	}
+		// TODO
+		// https://github.com/reactor/reactor-netty/issues/20
+		assumeFalse(client instanceof ReactorNettyWebSocketClient);
 
-	@Test
-	public void subProtocolUndertowClient() throws Exception {
-		if (server instanceof RxNettyHttpServer) {
-			// Caused by: java.io.IOException: Upgrade responses cannot have a transfer coding
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState.handleUpgrade(HttpUpgrade.java:490)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState.access$1200(HttpUpgrade.java:165)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:461)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:400)
-			// at org.xnio.ChannelListeners.invokeChannelListener(ChannelListeners.java:92)
-
-			return;
-		}
-		testSubProtocol(new UndertowWebSocketClient());
-	}
-
-	private void testSubProtocol(WebSocketClient client) throws URISyntaxException {
 		String protocol = "echo-v1";
 		AtomicReference<HandshakeInfo> infoRef = new AtomicReference<>();
 		MonoProcessor<Object> output = MonoProcessor.create();
@@ -188,45 +112,10 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
 	}
 
 	@Test
-	public void customHeaderReactorClient() throws Exception {
-		testCustomHeader(new ReactorNettyWebSocketClient());
-	}
-
-	@Test
-	public void customHeaderRxNettyClient() throws Exception {
-		testCustomHeader(new RxNettyWebSocketClient());
-	}
-
-	@Test
-	@Ignore
-	public void customHeaderJettyClient() throws Exception {
-		JettyWebSocketClient client = new JettyWebSocketClient();
-		client.start();
-		testCustomHeader(client);
-		client.stop();
-	}
-
-	@Test
-	public void customHeaderStandardClient() throws Exception {
-		testCustomHeader(new StandardWebSocketClient());
-	}
-
-	@Test
-	public void customHeaderUndertowClient() throws Exception {
-		if (server instanceof RxNettyHttpServer) {
-			// Caused by: java.io.IOException: Upgrade responses cannot have a transfer coding
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState.handleUpgrade(HttpUpgrade.java:490)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState.access$1200(HttpUpgrade.java:165)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:461)
-			// at org.xnio.http.HttpUpgrade$HttpUpgradeState$UpgradeResultListener.handleEvent(HttpUpgrade.java:400)
-			// at org.xnio.ChannelListeners.invokeChannelListener(ChannelListeners.java:92)
-
-			return;
-		}
-		testCustomHeader(new UndertowWebSocketClient());
-	}
+	public void customHeader() throws Exception {
 
-	private void testCustomHeader(WebSocketClient client) throws Exception {
+		// TODO
+		assumeFalse(client instanceof JettyWebSocketClient);
 
 		HttpHeaders headers = new HttpHeaders();
 		headers.add("my-header", "my-value");