Initial websockets tests.

Straight port from spring-web/tests.

TODO: add gateway support

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RemoveResponseHeaderWebFilterFactory.java

@@ -39,9 +39,8 @@ public List<String> argNames() {
 	public WebFilter apply(Tuple args) {
 		final String header = args.getString(NAME_KEY);
 
-		return (exchange, chain) -> chain.filter(exchange).then(Mono.defer(() -> {
+		return (exchange, chain) -> chain.filter(exchange).doFinally(v -> {
 			exchange.getResponse().getHeaders().remove(header);
-			return Mono.empty();
-		}));
+		});
 	}
 }

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/AbstractHttpServer.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2002-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.test.websocket;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.http.server.reactive.ContextPathCompositeHandler;
+import org.springframework.http.server.reactive.HttpHandler;
+import org.springframework.util.Assert;
+
+/**
+ * @author Rossen Stoyanchev
+ */
+public abstract class AbstractHttpServer implements HttpServer {
+
+	private String host = "0.0.0.0";
+
+	private int port = 0;
+
+	private HttpHandler httpHandler;
+
+	private Map<String, HttpHandler> handlerMap;
+
+	private boolean running;
+
+	private final Object lifecycleMonitor = new Object();
+
+
+	@Override
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public String getHost() {
+		return host;
+	}
+
+	@Override
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	@Override
+	public int getPort() {
+		return this.port;
+	}
+
+	@Override
+	public void setHandler(HttpHandler handler) {
+		this.httpHandler = handler;
+	}
+
+	public HttpHandler getHttpHandler() {
+		return this.httpHandler;
+	}
+
+	public void registerHttpHandler(String contextPath, HttpHandler handler) {
+		if (this.handlerMap == null) {
+			this.handlerMap = new LinkedHashMap<>();
+		}
+		this.handlerMap.put(contextPath, handler);
+	}
+
+	public Map<String, HttpHandler> getHttpHandlerMap() {
+		return this.handlerMap;
+	}
+
+	protected HttpHandler resolveHttpHandler() {
+		return getHttpHandlerMap() != null ?
+				new ContextPathCompositeHandler(getHttpHandlerMap()) : getHttpHandler();
+	}
+
+
+	// InitializingBean
+
+	@Override
+	public final void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.host, "Host must not be null");
+		Assert.isTrue(this.port >= 0, "Port must not be a negative number");
+		Assert.isTrue(this.httpHandler != null || this.handlerMap != null, "No HttpHandler configured");
+		Assert.state(!this.running, "Cannot reconfigure while running");
+
+		synchronized (this.lifecycleMonitor) {
+			initServer();
+		}
+	}
+
+	protected abstract void initServer() throws Exception;
+
+
+	// Lifecycle
+
+	@Override
+	public boolean isRunning() {
+		synchronized (this.lifecycleMonitor) {
+			return this.running;
+		}
+	}
+
+	@Override
+	public final void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!isRunning()) {
+				this.running = true;
+				try {
+					startInternal();
+				}
+				catch (Throwable ex) {
+					throw new IllegalStateException(ex);
+				}
+			}
+		}
+
+	}
+
+	protected abstract void startInternal() throws Exception;
+
+	@Override
+	public final void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (isRunning()) {
+				this.running = false;
+				try {
+					stopInternal();
+				}
+				catch (Throwable ex) {
+					throw new IllegalStateException(ex);
+				}
+				finally {
+					reset();
+				}
+			}
+		}
+	}
+
+	protected abstract void stopInternal() throws Exception;
+
+	private void reset() {
+		this.host = "0.0.0.0";
+		this.port = 0;
+		this.httpHandler = null;
+		this.handlerMap = null;
+		resetInternal();
+	}
+
+	protected abstract void resetInternal();
+
+}

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/HttpServer.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2002-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.test.websocket;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.Lifecycle;
+import org.springframework.http.server.reactive.HttpHandler;
+
+/**
+ * @author Rossen Stoyanchev
+ */
+public interface HttpServer extends InitializingBean, Lifecycle {
+
+	void setHost(String host);
+
+	void setPort(int port);
+
+	int getPort();
+
+	void setHandler(HttpHandler handler);
+
+}

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/ReactorHttpServer.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2002-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.test.websocket;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
+
+import reactor.ipc.netty.NettyContext;
+
+/**
+ * @author Stephane Maldini
+ */
+public class ReactorHttpServer extends AbstractHttpServer {
+
+	private ReactorHttpHandlerAdapter reactorHandler;
+
+	private reactor.ipc.netty.http.server.HttpServer reactorServer;
+
+	private AtomicReference<NettyContext> nettyContext = new AtomicReference<>();
+
+
+	@Override
+	protected void initServer() throws Exception {
+		this.reactorHandler = createHttpHandlerAdapter();
+		this.reactorServer = reactor.ipc.netty.http.server.HttpServer.create(getHost(), getPort());
+	}
+
+	private ReactorHttpHandlerAdapter createHttpHandlerAdapter() {
+		return new ReactorHttpHandlerAdapter(resolveHttpHandler());
+	}
+
+	@Override
+	protected void startInternal() {
+		NettyContext nettyContext = this.reactorServer.newHandler(this.reactorHandler).block();
+		setPort(nettyContext.address().getPort());
+		this.nettyContext.set(nettyContext);
+	}
+
+	@Override
+	protected void stopInternal() {
+		this.nettyContext.get().dispose();
+	}
+
+	@Override
+	protected void resetInternal() {
+		this.reactorServer = null;
+		this.reactorHandler = null;
+		this.nettyContext.set(null);
+	}
+
+}