Add @FunctionalSpringBootTest and mini web server

User can run a minimal HTTP app using an app that is a Function
or an ApplicationContextInitializer. Can also test using
@FunctionalSpringBootTest in place of @SpringBootTest.

Add some tests and documentation for functional beans

Make server.address configurable
This commit is contained in:
Dave Syer
2018-10-18 16:06:50 +01:00
committed by Oleg Zhurakousky
parent 4315cb1d61
commit ba34d4b81b
16 changed files with 703 additions and 22 deletions

View File

@@ -0,0 +1,202 @@
package org.springframework.cloud.function.web.function;
import java.lang.management.ManagementFactory;
import java.time.Duration;
import java.util.Set;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.boot.autoconfigure.web.ErrorProperties;
import org.springframework.boot.autoconfigure.web.ResourceProperties;
import org.springframework.boot.autoconfigure.web.reactive.error.DefaultErrorWebExceptionHandler;
import org.springframework.boot.web.reactive.error.DefaultErrorAttributes;
import org.springframework.boot.web.reactive.error.ErrorAttributes;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.SmartApplicationListener;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.util.ClassUtils;
import org.springframework.web.reactive.function.server.HandlerStrategies;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.server.WebExceptionHandler;
import org.springframework.web.server.adapter.HttpWebHandlerAdapter;
import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
import static org.springframework.web.reactive.function.server.RequestPredicates.POST;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;
import static org.springframework.web.reactive.function.server.ServerResponse.ok;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;
public class FunctionEndpointInitializer
implements ApplicationContextInitializer<GenericApplicationContext> {
@Override
public void initialize(GenericApplicationContext context) {
if (context.getEnvironment().getProperty("spring.functional.enabled",
Boolean.class, false) && ClassUtils.isPresent("org.springframework.http.server.reactive.HttpHandler", null)) {
registerEndpoint(context);
registerWebFluxAutoConfiguration(context);
}
}
private void registerWebFluxAutoConfiguration(GenericApplicationContext context) {
context.registerBean(DefaultErrorWebExceptionHandler.class,
() -> errorHandler(context));
context.registerBean(WebHttpHandlerBuilder.WEB_HANDLER_BEAN_NAME,
HttpWebHandlerAdapter.class, () -> httpHandler(context));
context.addApplicationListener(new ServerListener(context));
}
private void registerEndpoint(GenericApplicationContext context) {
context.registerBean(FunctionEndpointFactory.class,
() -> new FunctionEndpointFactory(context.getBean(FunctionCatalog.class),
context.getBean(FunctionInspector.class),
context.getEnvironment()));
context.registerBean(RouterFunction.class,
() -> context.getBean(FunctionEndpointFactory.class).functionEndpoints());
}
private HttpWebHandlerAdapter httpHandler(GenericApplicationContext context) {
return (HttpWebHandlerAdapter) RouterFunctions.toHttpHandler(
context.getBean(RouterFunction.class),
HandlerStrategies.empty()
.exceptionHandler(context.getBean(WebExceptionHandler.class))
.codecs(config -> config.registerDefaults(true)).build());
}
private DefaultErrorWebExceptionHandler errorHandler(
GenericApplicationContext context) {
context.registerBean(ErrorAttributes.class, () -> new DefaultErrorAttributes());
context.registerBean(ErrorProperties.class, () -> new ErrorProperties());
context.registerBean(ResourceProperties.class, () -> new ResourceProperties());
DefaultErrorWebExceptionHandler handler = new DefaultErrorWebExceptionHandler(
context.getBean(ErrorAttributes.class),
context.getBean(ResourceProperties.class),
context.getBean(ErrorProperties.class), context);
ServerCodecConfigurer codecs = ServerCodecConfigurer.create();
handler.setMessageWriters(codecs.getWriters());
handler.setMessageReaders(codecs.getReaders());
return handler;
}
private static class ServerListener implements SmartApplicationListener {
private static Log logger = LogFactory.getLog(ServerListener.class);
private GenericApplicationContext context;
public ServerListener(GenericApplicationContext context) {
this.context = context;
}
@Override
public void onApplicationEvent(ApplicationEvent event) {
ApplicationContext context = ((ContextRefreshedEvent) event)
.getApplicationContext();
if (context != this.context) {
return;
}
if (!ClassUtils.isPresent(
"org.springframework.http.server.reactive.HttpHandler", null)) {
logger.info("No web server classes found so no server to start");
return;
}
Integer port = Integer.valueOf(context.getEnvironment()
.resolvePlaceholders("${server.port:${PORT:8080}}"));
String address = context.getEnvironment()
.resolvePlaceholders("${server.address:0.0.0.0}");
if (port >= 0) {
HttpHandler handler = context.getBean(HttpHandler.class);
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(
handler);
HttpServer httpServer = HttpServer.create().host(address).port(port)
.handle(adapter);
Thread thread = new Thread(() -> httpServer
.bindUntilJavaShutdown(Duration.ofSeconds(60), this::callback),
"server-startup");
thread.setDaemon(false);
thread.start();
}
}
private void callback(DisposableServer server) {
logger.info("Server started");
try {
double uptime = ManagementFactory.getRuntimeMXBean().getUptime();
System.err.println("JVM running for " + uptime + "ms");
}
catch (Throwable e) {
}
}
@Override
public boolean supportsEventType(Class<? extends ApplicationEvent> eventType) {
return eventType.isAssignableFrom(ContextRefreshedEvent.class);
}
}
}
class FunctionEndpointFactory {
private static Log logger = LogFactory.getLog(FunctionEndpointFactory.class);
private Function<Flux<?>, Flux<?>> function;
private FunctionInspector inspector;
public FunctionEndpointFactory(FunctionCatalog catalog, FunctionInspector inspector,
Environment environment) {
String handler = environment.resolvePlaceholders("${function.handler}");
if (handler.startsWith("$")) {
handler = null;
}
this.inspector = inspector;
this.function = extract(catalog, handler);
}
private Function<Flux<?>, Flux<?>> extract(FunctionCatalog catalog, String handler) {
Set<String> names = catalog.getNames(Function.class);
if (!names.isEmpty()) {
logger.info("Found functions: " + names);
if (handler != null) {
logger.info("Configured function: " + handler);
if (!names.contains(handler)) {
throw new IllegalStateException("Cannot locate function: " + handler);
}
return catalog.lookup(Function.class, handler);
}
return catalog.lookup(Function.class, names.iterator().next());
}
throw new IllegalStateException("No function defined");
}
@SuppressWarnings({ "unchecked" })
public <T> RouterFunction<?> functionEndpoints() {
return route(POST("/"), request -> {
Class<?> inputType = this.inspector.getInputType(this.function);
Class<T> outputType = (Class<T>) this.inspector.getOutputType(this.function);
return ok().body(
Mono.from(
(Flux<T>) this.function.apply(request.bodyToFlux(inputType))),
outputType);
});
}
}

View File

@@ -6,3 +6,6 @@ org.springframework.cloud.function.web.source.SupplierAutoConfiguration
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc=\
org.springframework.cloud.function.web.flux.ReactorAutoConfiguration,\
org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration
org.springframework.context.ApplicationContextInitializer=\
org.springframework.cloud.function.web.function.FunctionEndpointInitializer