Update to latest reactor-netty stack

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestConfiguration.java

@@ -16,13 +16,6 @@
 
 package org.springframework.cloud.function.web;
 
-import static org.springframework.http.MediaType.TEXT_PLAIN;
-import static org.springframework.web.reactive.function.BodyExtractors.toFlux;
-import static org.springframework.web.reactive.function.BodyInserters.fromPublisher;
-import static org.springframework.web.reactive.function.RequestPredicates.POST;
-import static org.springframework.web.reactive.function.RequestPredicates.contentType;
-
-import java.util.concurrent.Executors;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
@@ -35,22 +28,29 @@ import org.springframework.cloud.function.registry.FileSystemFunctionRegistry;
 import org.springframework.cloud.function.registry.FunctionRegistry;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.http.MediaType;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
 import org.springframework.util.StringUtils;
-import org.springframework.web.reactive.function.Request;
-import org.springframework.web.reactive.function.Response;
+import org.springframework.web.reactive.function.RequestPredicates;
 import org.springframework.web.reactive.function.RouterFunction;
 import org.springframework.web.reactive.function.RouterFunctions;
+import org.springframework.web.reactive.function.ServerRequest;
+import org.springframework.web.reactive.function.ServerResponse;
+
+import static org.springframework.http.codec.BodyExtractors.toFlux;
+import static org.springframework.http.codec.BodyInserters.fromPublisher;
 
 import reactor.core.publisher.Flux;
-import reactor.ipc.netty.http.HttpServer;
+import reactor.ipc.netty.NettyContext;
+import reactor.ipc.netty.http.server.HttpServer;
 
 /**
  * @author Mark Fisher
  */
 @Configuration
-@EnableConfigurationProperties({ FunctionConfigurationProperties.class, WebConfigurationProperties.class })
+@EnableConfigurationProperties({ FunctionConfigurationProperties.class,
+		WebConfigurationProperties.class })
 public class RestConfiguration {
 
 	@Autowired
@@ -69,10 +69,13 @@ public class RestConfiguration {
 		String name = functionProperties.getName();
 		Function<Flux<String>, Flux<String>> function = (name.indexOf(',') == -1)
 				? registry.lookupFunction(name)
-				: registry.composeFunction(StringUtils.commaDelimitedListToStringArray(name));
+				: registry.composeFunction(
+						StringUtils.commaDelimitedListToStringArray(name));
 		FunctionInvokingHandler handler = new FunctionInvokingHandler(function);
 		RouterFunction<Publisher<String>> route = RouterFunctions.route(
-				POST(webProperties.getPath()).and(contentType(TEXT_PLAIN)), handler::handleText);
+				RequestPredicates.POST(webProperties.getPath())
+						.and(RequestPredicates.contentType(MediaType.TEXT_PLAIN)),
+				handler::handleText);
 		return RouterFunctions.toHttpHandler(route);
 	}
 
@@ -89,20 +92,21 @@ public class RestConfiguration {
 			this.function = function;
 		}
 
-		private  Response<Publisher<String>> handleText(Request request) {
+		private ServerResponse<Publisher<String>> handleText(ServerRequest request) {
 			Flux<String> input = request.body(toFlux(String.class));
 			Publisher<String> output = this.function.apply(input);
-			return Response.ok().body(fromPublisher(output, String.class));
+			return ServerResponse.ok().body(fromPublisher(output, String.class));
 		}
 	}
 
-	private static class LifecycleAwareHttpServer implements InitializingBean, DisposableBean {
+	private static class LifecycleAwareHttpServer
+			implements InitializingBean, DisposableBean {
 
 		private final HttpHandler handler;
 
 		private final int port;
 
-		private volatile HttpServer server;
+		private volatile NettyContext server;
 
 		private LifecycleAwareHttpServer(HttpHandler handler, int port) {
 			this.handler = handler;
@@ -111,27 +115,34 @@ public class RestConfiguration {
 
 		@Override
 		public void afterPropertiesSet() {
-			ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(this.handler);
+			ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(
+					this.handler);
 			final HttpServer server = HttpServer.create("localhost", port);
-			this.server = server;
-			Executors.newSingleThreadExecutor().submit(new Runnable() {
+			Thread thread = new Thread(new Runnable() {
 
 				@Override
 				public void run() {
-					try {
-						server.startAndAwait(adapter);
-					}
-					catch (InterruptedException e) {
-						Thread.currentThread().interrupt();
+					LifecycleAwareHttpServer.this.server = server.newHandler(adapter)
+							.block();
+					while (LifecycleAwareHttpServer.this.server != null) {
+						try {
+							Thread.sleep(100L);
+						}
+						catch (InterruptedException e) {
+							Thread.currentThread().interrupt();
+						}
 					}
 				}
 			});
+			thread.setDaemon(false);
+			thread.start();
 		}
 
 		@Override
 		public void destroy() {
 			if (this.server != null) {
-				this.server.shutdown();
+				this.server.dispose();
+				this.server = null;
 			}
 		}
 	}