diff --git a/pom.xml b/pom.xml index eb71b1df5..d936aa831 100644 --- a/pom.xml +++ b/pom.xml @@ -56,6 +56,7 @@ spring-cloud-function-context spring-cloud-function-stream spring-cloud-function-task + spring-cloud-function-web spring-cloud-function-samples spring-cloud-function-deployer spring-cloud-function-adapters diff --git a/spring-cloud-function-dependencies/pom.xml b/spring-cloud-function-dependencies/pom.xml index 414ba1958..ad6010252 100644 --- a/spring-cloud-function-dependencies/pom.xml +++ b/spring-cloud-function-dependencies/pom.xml @@ -70,6 +70,11 @@ spring-cloud-function-adapter-openwhisk ${project.version} + + org.springframework.cloud + spring-cloud-stream-binder-servlet + ${project.version} + diff --git a/spring-cloud-function-deployer/pom.xml b/spring-cloud-function-deployer/pom.xml index a5492101e..e5e25c304 100644 --- a/spring-cloud-function-deployer/pom.xml +++ b/spring-cloud-function-deployer/pom.xml @@ -15,8 +15,7 @@ - 1.0.8.RELEASE - 1.0.0.BUILD-SNAPSHOT + 1.0.9.RELEASE @@ -31,7 +30,6 @@ org.springframework.cloud spring-cloud-stream-binder-servlet - ${spring-cloud-stream-servlet.version} org.springframework.boot diff --git a/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/ApplicationRunner.java b/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/ApplicationRunner.java index c603aebf9..872d1ab72 100644 --- a/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/ApplicationRunner.java +++ b/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/ApplicationRunner.java @@ -99,6 +99,7 @@ public class ApplicationRunner implements CommandLineRunner { if (this.classLoader!=null && this.classLoader instanceof Closeable) { ((Closeable) this.classLoader).close(); } + this.classLoader = null; } private RuntimeException getError() { diff --git a/spring-cloud-function-web/.jdk8 b/spring-cloud-function-web/.jdk8 new file mode 100644 index 000000000..e69de29bb diff --git a/spring-cloud-function-web/pom.xml b/spring-cloud-function-web/pom.xml new file mode 100644 index 000000000..259d217cb --- /dev/null +++ b/spring-cloud-function-web/pom.xml @@ -0,0 +1,50 @@ + + + 4.0.0 + + spring-cloud-function-web + jar + Spring Cloud Function Web Support + Spring Cloud Function Web Support + + + org.springframework.cloud + spring-cloud-function-parent + 1.0.0.BUILD-SNAPSHOT + + + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.cloud + spring-cloud-function-context + ${project.version} + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.springframework.boot.experimental + spring-boot-thin-layout + ${wrapper.version} + + + + + + + diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestApplication.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestApplication.java new file mode 100644 index 000000000..f59f2afe5 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestApplication.java @@ -0,0 +1,32 @@ +/* + * Copyright 2016-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.function.web; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * @author Mark Fisher + */ +@SpringBootApplication +public class RestApplication { + + public static void main(String[] args) { + SpringApplication.run(RestApplication.class, args); + } +} + diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java new file mode 100644 index 000000000..b12816670 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java @@ -0,0 +1,123 @@ +/* + * Copyright 2016-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.function.web.flux; + +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.web.flux.request.FluxRequest; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestAttribute; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.ResponseBody; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * @author Dave Syer + * @author Mark Fisher + */ +@Component +public class FunctionController { + + private static Log logger = LogFactory.getLog(FunctionController.class); + + private FunctionInspector inspector; + + private boolean debug = false; + + public FunctionController(FunctionInspector inspector) { + this.inspector = inspector; + } + + public void setDebug(boolean debug) { + this.debug = debug; + } + + @PostMapping(path = "/**") + @ResponseBody + public ResponseEntity> post( + @RequestAttribute(required = false, name = "org.springframework.cloud.function.web.flux.constants.WebRequestConstants.function") Function, Flux> function, + @RequestAttribute(required = false, name = "org.springframework.cloud.function.web.flux.constants.WebRequestConstants.consumer") Consumer> consumer, + @RequestAttribute(required = false, name = "org.springframework.cloud.function.web.flux.constants.WebRequestConstants.input_single") Boolean single, + @RequestBody FluxRequest body) { + if (function != null) { + Flux flux = body.flux(); + if (debug) { + flux = flux.log(); + } + Flux result = function.apply(flux); + if (logger.isDebugEnabled()) { + logger.debug("Handled POST with function"); + } + return ResponseEntity.ok().body(debug ? result.log() : result); + } + if (consumer != null) { + Flux flux = body.flux().cache(); // send a copy back to the caller + if (debug) { + flux = flux.log(); + } + consumer.accept(flux); + if (logger.isDebugEnabled()) { + logger.debug("Handled POST with consumer"); + } + return ResponseEntity.status(HttpStatus.ACCEPTED).body(flux); + } + throw new IllegalArgumentException("no such function"); + } + + @GetMapping(path = "/**") + @ResponseBody + public Object get( + @RequestAttribute(required = false, name = "org.springframework.cloud.function.web.flux.constants.WebRequestConstants.function") Function, Flux> function, + @RequestAttribute(required = false, name = "org.springframework.cloud.function.web.flux.constants.WebRequestConstants.supplier") Supplier> supplier, + @RequestAttribute(required = false, name = "org.springframework.cloud.function.web.flux.constants.WebRequestConstants.argument") String argument) { + if (function != null) { + return value(function, argument); + } + return supplier(supplier); + } + + private Flux supplier(Supplier> supplier) { + Flux result = supplier.get(); + if (logger.isDebugEnabled()) { + logger.debug("Handled GET with supplier"); + } + return debug ? result.log() : result; + } + + private Mono value(Function, Flux> function, + @PathVariable String value) { + Object input = inspector.convert(function, value); + Mono result = Mono.from(function.apply(Flux.just(input))); + if (logger.isDebugEnabled()) { + logger.debug("Handled GET with function"); + } + return debug ? result.log() : result; + } +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionHandlerMapping.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionHandlerMapping.java new file mode 100644 index 000000000..367806968 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionHandlerMapping.java @@ -0,0 +1,165 @@ +/* + * Copyright 2012-2015 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.function.web.flux; + +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import javax.servlet.http.HttpServletRequest; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.core.FunctionCatalog; +import org.springframework.cloud.function.web.flux.constants.WebRequestConstants; +import org.springframework.context.annotation.Configuration; +import org.springframework.util.StringUtils; +import org.springframework.web.method.HandlerMethod; +import org.springframework.web.servlet.HandlerMapping; +import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping; + +/** + * @author Dave Syer + * + */ +@Configuration +@ConditionalOnClass(RequestMappingHandlerMapping.class) +public class FunctionHandlerMapping extends RequestMappingHandlerMapping + implements InitializingBean { + + private final FunctionCatalog functions; + + private final FunctionController controller; + + @Value("${spring.cloud.function.web.path:}") + private String prefix = ""; + + @Value("${debug:${DEBUG:false}}") + private String debug = "false"; + + @Autowired + public FunctionHandlerMapping(FunctionCatalog catalog, FunctionInspector inspector) { + this.functions = catalog; + logger.info("FunctionCatalog: " + catalog + ", FunctionInspector: " + inspector); + setOrder(super.getOrder() - 5); + this.controller = new FunctionController(inspector); + } + + @Override + public void afterPropertiesSet() { + super.afterPropertiesSet(); + this.controller.setDebug(!"false".equals(debug)); + detectHandlerMethods(controller); + while (prefix.endsWith("/")) { + prefix = prefix.substring(0, prefix.length() - 1); + } + } + + @Override + protected void initHandlerMethods() { + } + + @Override + protected HandlerMethod getHandlerInternal(HttpServletRequest request) + throws Exception { + HandlerMethod handler = super.getHandlerInternal(request); + if (handler == null) { + return null; + } + String path = (String) request + .getAttribute(HandlerMapping.PATH_WITHIN_HANDLER_MAPPING_ATTRIBUTE); + if (StringUtils.hasText(prefix) && !path.startsWith(prefix)) { + return null; + } + if (path.startsWith(prefix)) { + path = path.substring(prefix.length()); + } + if (path == null) { + return handler; + } + Object function = findFunctionForGet(request, path); + if (function != null) { + if (logger.isDebugEnabled()) { + logger.debug("Found function for GET: " + path); + } + request.setAttribute(WebRequestConstants.HANDLER, function); + return handler; + } + function = findFunctionForPost(request, path); + if (function != null) { + if (logger.isDebugEnabled()) { + logger.debug("Found function for POST: " + path); + } + request.setAttribute(WebRequestConstants.HANDLER, function); + return handler; + } + return null; + } + + private Object findFunctionForPost(HttpServletRequest request, String path) { + if (!request.getMethod().equals("POST")) { + return null; + } + path = path.startsWith("/") ? path.substring(1) : path; + Consumer consumer = functions.lookupConsumer(path); + if (consumer != null) { + request.setAttribute(WebRequestConstants.CONSUMER, consumer); + return consumer; + } + Function function = functions.lookupFunction(path); + if (function != null) { + request.setAttribute(WebRequestConstants.FUNCTION, function); + return function; + } + return null; + } + + private Object findFunctionForGet(HttpServletRequest request, String path) { + if (!request.getMethod().equals("GET")) { + return null; + } + path = path.startsWith("/") ? path.substring(1) : path; + Supplier supplier = functions.lookupSupplier(path); + if (supplier != null) { + request.setAttribute(WebRequestConstants.SUPPLIER, supplier); + return supplier; + } + StringBuilder builder = new StringBuilder(); + String name = path; + String value = null; + for (String element : path.split("/")) { + if (builder.length() > 0) { + builder.append("/"); + } + builder.append(element); + name = builder.toString(); + value = path.length() > name.length() ? path.substring(name.length() + 1) + : null; + Function function = functions.lookupFunction(name); + if (function != null) { + request.setAttribute(WebRequestConstants.FUNCTION, function); + request.setAttribute(WebRequestConstants.ARGUMENT, value); + return function; + } + } + return null; + } + +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/ReactorAutoConfiguration.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/ReactorAutoConfiguration.java new file mode 100644 index 000000000..d5394b1b7 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/ReactorAutoConfiguration.java @@ -0,0 +1,114 @@ +/* + * Copyright 2013-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.function.web.flux; + +import java.util.ArrayList; +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication; +import org.springframework.boot.autoconfigure.web.HttpMessageConverters; +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.core.FunctionCatalog; +import org.springframework.cloud.function.web.flux.request.FluxHandlerMethodArgumentResolver; +import org.springframework.cloud.function.web.flux.response.FluxReturnValueHandler; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.util.ClassUtils; +import org.springframework.web.method.support.AsyncHandlerMethodReturnValueHandler; +import org.springframework.web.method.support.HandlerMethodArgumentResolver; +import org.springframework.web.method.support.HandlerMethodReturnValueHandler; +import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * @author Mark Fisher + */ +@Configuration +@ConditionalOnWebApplication +@ConditionalOnClass({ Flux.class, AsyncHandlerMethodReturnValueHandler.class }) +public class ReactorAutoConfiguration { + + @Autowired + private ApplicationContext context; + + @Bean + public FunctionHandlerMapping functionHandlerMapping(FunctionCatalog catalog, + FunctionInspector inspector) { + return new FunctionHandlerMapping(catalog, inspector); + } + + @Configuration + @ConditionalOnMissingClass("org.springframework.core.ReactiveAdapter") + protected static class FluxReturnValueConfiguration { + @Bean + public FluxReturnValueHandler fluxReturnValueHandler(FunctionInspector inspector, + HttpMessageConverters converters) { + return new FluxReturnValueHandler(inspector, converters.getConverters()); + } + } + + @Configuration + protected static class FluxArgumentResolverConfiguration { + @Bean + public FluxHandlerMethodArgumentResolver fluxHandlerMethodArgumentResolver( + FunctionInspector inspector, ObjectMapper mapper) { + return new FluxHandlerMethodArgumentResolver(inspector, mapper); + } + } + + @Bean + public BeanPostProcessor fluxRequestMappingHandlerAdapterProcessor() { + return new BeanPostProcessor() { + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) + throws BeansException { + if (bean instanceof RequestMappingHandlerAdapter) { + RequestMappingHandlerAdapter adapter = (RequestMappingHandlerAdapter) bean; + List resolvers = new ArrayList<>( + adapter.getArgumentResolvers()); + resolvers.add(0, + context.getBean(FluxHandlerMethodArgumentResolver.class)); + adapter.setArgumentResolvers(resolvers); + if (!ClassUtils.isPresent("org.springframework.core.ReactiveAdapter", + null)) { + List handlers = new ArrayList<>( + adapter.getReturnValueHandlers()); + handlers.add(0, context.getBean(FluxReturnValueHandler.class)); + adapter.setReturnValueHandlers(handlers); + } + } + return bean; + } + + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) + throws BeansException { + return bean; + } + }; + } +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/constants/WebRequestConstants.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/constants/WebRequestConstants.java new file mode 100644 index 000000000..720313dbb --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/constants/WebRequestConstants.java @@ -0,0 +1,39 @@ +/* + * Copyright 2016-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.function.web.flux.constants; + +/** + * Common storage for web request attribute names (in a separate package to avoid cycles). + * + * @author Dave Syer + * + */ +public abstract class WebRequestConstants { + + public static final String FUNCTION = WebRequestConstants.class.getName() + + ".function"; + public static final String CONSUMER = WebRequestConstants.class.getName() + + ".consumer"; + public static final String SUPPLIER = WebRequestConstants.class.getName() + + ".supplier"; + public static final String ARGUMENT = WebRequestConstants.class.getName() + + ".argument"; + public static final String HANDLER = WebRequestConstants.class.getName() + ".handler"; + public static final String INPUT_SINGLE = WebRequestConstants.class.getName() + + ".input_single"; + +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/DelegateHandler.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/DelegateHandler.java new file mode 100644 index 000000000..1749d2580 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/DelegateHandler.java @@ -0,0 +1,53 @@ +/* + * Copyright 2016-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.function.web.flux.request; + +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.cloud.function.context.catalog.FunctionInspector; + +public abstract class DelegateHandler { + + private final ListableBeanFactory factory; + private FunctionInspector processor; + private Object handler; + private final Object source; + + public DelegateHandler(ListableBeanFactory factory, Object source) { + this.factory = factory; + this.source = source; + } + + public Class type() { + return processor().getInputType(handler()); + } + + private Object handler() { + if (handler == null) { + handler = source instanceof String ? factory.getBean((String) source) + : source; + } + return handler; + } + + private FunctionInspector processor() { + if (processor == null) { + processor = factory.getBean(FunctionInspector.class); + } + return processor; + } + +} \ No newline at end of file diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxHandlerMethodArgumentResolver.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxHandlerMethodArgumentResolver.java new file mode 100644 index 000000000..c7e60e485 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxHandlerMethodArgumentResolver.java @@ -0,0 +1,134 @@ +/* + * Copyright 2016-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.function.web.flux.request; + +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import javax.servlet.http.HttpServletRequest; + +import com.fasterxml.jackson.databind.JsonMappingException; +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.web.flux.constants.WebRequestConstants; +import org.springframework.cloud.function.web.util.HeaderUtils; +import org.springframework.core.MethodParameter; +import org.springframework.core.Ordered; +import org.springframework.http.MediaType; +import org.springframework.http.server.ServletServerHttpRequest; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.StreamUtils; +import org.springframework.web.bind.support.WebDataBinderFactory; +import org.springframework.web.context.request.NativeWebRequest; +import org.springframework.web.method.support.HandlerMethodArgumentResolver; +import org.springframework.web.method.support.ModelAndViewContainer; +import org.springframework.web.util.ContentCachingRequestWrapper; + +/** + * Converter for request bodies of type Flux. + * + * @author Dave Syer + * + */ +public class FluxHandlerMethodArgumentResolver + implements HandlerMethodArgumentResolver, Ordered { + + private static Log logger = LogFactory + .getLog(FluxHandlerMethodArgumentResolver.class); + + private final ObjectMapper mapper; + + private FunctionInspector inspector; + + public FluxHandlerMethodArgumentResolver(FunctionInspector inspector, + ObjectMapper mapper) { + this.inspector = inspector; + this.mapper = mapper; + } + + @Override + public int getOrder() { + return Ordered.HIGHEST_PRECEDENCE; + } + + @Override + public Object resolveArgument(MethodParameter parameter, + ModelAndViewContainer mavContainer, NativeWebRequest webRequest, + WebDataBinderFactory binderFactory) throws Exception { + Object handler = webRequest.getAttribute(WebRequestConstants.HANDLER, + NativeWebRequest.SCOPE_REQUEST); + Class type = inspector.getInputType(handler); + if (type == null) { + type = Object.class; + } + boolean message = inspector.isMessage(handler); + List body; + ContentCachingRequestWrapper nativeRequest = new ContentCachingRequestWrapper( + webRequest.getNativeRequest(HttpServletRequest.class)); + if (logger.isDebugEnabled()) { + logger.debug("Resolving request body into type: " + type); + } + if (isPlainText(webRequest) && CharSequence.class.isAssignableFrom(type)) { + body = Arrays.asList(StreamUtils.copyToString(nativeRequest.getInputStream(), + Charset.forName("UTF-8"))); + } + else { + try { + body = mapper.readValue(nativeRequest.getInputStream(), + mapper.getTypeFactory() + .constructCollectionLikeType(ArrayList.class, type)); + } + catch (JsonMappingException e) { + nativeRequest.setAttribute(WebRequestConstants.INPUT_SINGLE, true); + body = Arrays.asList( + mapper.readValue(nativeRequest.getContentAsByteArray(), type)); + } + } + if (message) { + List messages = new ArrayList<>(); + for (Object payload : body) { + messages.add(MessageBuilder.withPayload(payload) + .copyHeaders(HeaderUtils.fromHttp(new ServletServerHttpRequest( + webRequest.getNativeRequest(HttpServletRequest.class)) + .getHeaders())) + .build()); + } + body = messages; + } + return new FluxRequest(body); + } + + private boolean isPlainText(NativeWebRequest webRequest) { + String value = webRequest.getHeader("Content-Type"); + if (value != null) { + return MediaType.valueOf(value).isCompatibleWith(MediaType.TEXT_PLAIN); + } + return false; + } + + @Override + public boolean supportsParameter(MethodParameter parameter) { + return FluxRequest.class.isAssignableFrom(parameter.getParameterType()); + } + +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxRequest.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxRequest.java new file mode 100644 index 000000000..00b2225fc --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxRequest.java @@ -0,0 +1,45 @@ +/* + * Copyright 2016-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.function.web.flux.request; + +import java.util.List; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * + */ +public class FluxRequest { + + private List body; + + public FluxRequest(List body) { + this.body = body; + } + + public Flux flux() { + return Flux.fromIterable(body); + } + + public List body() { + return body; + } + +} + + diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxResponseBodyEmitter.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxResponseBodyEmitter.java new file mode 100644 index 000000000..d8814cca9 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxResponseBodyEmitter.java @@ -0,0 +1,60 @@ +/* + * Copyright 2013-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.function.web.flux.response; + +import org.reactivestreams.Publisher; + +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.server.ServerHttpResponse; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; + +import reactor.core.publisher.Flux; + +/** + * A specialized {@link ResponseBodyEmitter} that handles {@link Flux} return types. + * + * @author Dave Syer + */ +class FluxResponseBodyEmitter extends ResponseBodyEmitter { + + private final MediaType mediaType; + private ResponseBodyEmitterSubscriber subscriber; + + public FluxResponseBodyEmitter(Publisher observable) { + this(new HttpHeaders(), null, observable); + } + + public FluxResponseBodyEmitter(HttpHeaders request, MediaType mediaType, + Publisher observable) { + super(); + this.mediaType = mediaType; + this.subscriber = new ResponseBodyEmitterSubscriber(request, mediaType, + observable, this, MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)); + } + + @Override + protected void extendResponse(ServerHttpResponse outputMessage) { + super.extendResponse(outputMessage); + this.subscriber.extendResponse(outputMessage); + HttpHeaders headers = outputMessage.getHeaders(); + if (headers.getContentType() == null && this.mediaType != null + && !MediaType.ALL.equals(this.mediaType)) { + headers.setContentType(this.mediaType); + } + } +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxResponseSseEmitter.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxResponseSseEmitter.java new file mode 100644 index 000000000..07d9e0c5b --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxResponseSseEmitter.java @@ -0,0 +1,55 @@ +/* + * Copyright 2013-2016 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.function.web.flux.response; + +import org.reactivestreams.Publisher; + +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.server.ServerHttpResponse; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import reactor.core.publisher.Flux; + +/** + * A specialized {@link ResponseBodyEmitter} that handles {@link Flux} return types with + * SSE streams. + * + * @author Dave Syer + */ +class FluxResponseSseEmitter extends SseEmitter { + + private ResponseBodyEmitterSubscriber subscriber; + + public FluxResponseSseEmitter(Publisher observable) { + this(new HttpHeaders(), MediaType.valueOf("text/plain"), observable); + } + + public FluxResponseSseEmitter(HttpHeaders request, MediaType mediaType, + Publisher observable) { + super(); + this.subscriber = new ResponseBodyEmitterSubscriber(request, mediaType, + observable, this, false); + } + + @Override + protected void extendResponse(ServerHttpResponse outputMessage) { + super.extendResponse(outputMessage); + this.subscriber.extendResponse(outputMessage); + } +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxReturnValueHandler.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxReturnValueHandler.java new file mode 100644 index 000000000..a54793a52 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxReturnValueHandler.java @@ -0,0 +1,267 @@ +/* + * Copyright 2013-2016 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.function.web.flux.response; + +import java.lang.reflect.Method; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.function.Supplier; +import java.util.stream.Stream; + +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.el.stream.Optional; +import org.reactivestreams.Publisher; + +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.web.flux.constants.WebRequestConstants; +import org.springframework.cloud.function.web.util.HeaderUtils; +import org.springframework.core.MethodParameter; +import org.springframework.core.ResolvableType; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.http.server.ServletServerHttpRequest; +import org.springframework.messaging.Message; +import org.springframework.util.ReflectionUtils; +import org.springframework.web.context.request.NativeWebRequest; +import org.springframework.web.method.support.AsyncHandlerMethodReturnValueHandler; +import org.springframework.web.method.support.ModelAndViewContainer; +import org.springframework.web.servlet.mvc.method.annotation.RequestResponseBodyMethodProcessor; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitterReturnValueHandler; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * A specialized {@link AsyncHandlerMethodReturnValueHandler} that handles {@link Flux} + * return types. + * + * @author Dave Syer + */ +public class FluxReturnValueHandler implements AsyncHandlerMethodReturnValueHandler { + + private static Log logger = LogFactory.getLog(FluxReturnValueHandler.class); + + private ResponseBodyEmitterReturnValueHandler delegate; + private RequestResponseBodyMethodProcessor single; + private long timeout = 1000L; + private static final MediaType EVENT_STREAM = MediaType.valueOf("text/event-stream"); + + private FunctionInspector inspector; + + private MethodParameter singleReturnType; + + public FluxReturnValueHandler(FunctionInspector inspector, + List> messageConverters) { + this.inspector = inspector; + this.delegate = new ResponseBodyEmitterReturnValueHandler(messageConverters); + this.single = new RequestResponseBodyMethodProcessor(messageConverters); + Method method = ReflectionUtils.findMethod(getClass(), "singleValue"); + singleReturnType = new MethodParameter(method, -1); + } + + ResponseEntity singleValue() { + return null; + } + + /** + * Timeout for clients. If no items are seen on an HTTP response in this period then + * the response is closed. + * + * @param timeout the timeout to set + */ + public void setTimeout(long timeout) { + this.timeout = timeout; + } + + @Override + public boolean isAsyncReturnValue(Object returnValue, MethodParameter returnType) { + if (returnValue != null) { + return supportsReturnType(returnType); + } + return false; + } + + @Override + public boolean supportsReturnType(MethodParameter returnType) { + return (returnType.getParameterType() != null + && (Publisher.class.isAssignableFrom(returnType.getParameterType()) + || isResponseEntity(returnType))) + || Publisher.class + .isAssignableFrom(returnType.getMethod().getReturnType()); + } + + private boolean isResponseEntity(MethodParameter returnType) { + if (ResponseEntity.class.isAssignableFrom(returnType.getParameterType())) { + Class bodyType = ResolvableType.forMethodParameter(returnType) + .getGeneric(0).resolve(); + return bodyType != null && Flux.class.isAssignableFrom(bodyType); + } + return false; + } + + @Override + public void handleReturnValue(Object returnValue, MethodParameter returnType, + ModelAndViewContainer mavContainer, NativeWebRequest webRequest) + throws Exception { + + if (returnValue == null) { + mavContainer.setRequestHandled(true); + return; + } + + Object adaptFrom = returnValue; + if (returnValue instanceof ResponseEntity) { + ResponseEntity value = (ResponseEntity) returnValue; + adaptFrom = value.getBody(); + HttpServletResponse response = webRequest + .getNativeResponse(HttpServletResponse.class); + response.setStatus(value.getStatusCodeValue()); + HttpHeaders headers = value.getHeaders(); + for (String name : headers.keySet()) { + List list = headers.get(name); + for (String header : list) { + response.addHeader(name, header); + } + } + } + Publisher flux = (Publisher) adaptFrom; + + Object handler = webRequest.getAttribute(WebRequestConstants.HANDLER, + NativeWebRequest.SCOPE_REQUEST); + Class type = inspector.getOutputType(handler); + + boolean inputSingle = isInputSingle(webRequest, handler); + if (inputSingle && isOutputSingle(handler)) { + Object result = Flux.from(flux).blockFirst(); + if (result instanceof Message) { + Message message = (Message) result; + result = message.getPayload(); + addHeaders(webRequest, message); + } + single.handleReturnValue(result, singleReturnType, mavContainer, webRequest); + return; + } + + MediaType mediaType = null; + if (isPlainText(webRequest) && CharSequence.class.isAssignableFrom(type)) { + mediaType = MediaType.TEXT_PLAIN; + } + else { + mediaType = findMediaType(webRequest); + } + if (logger.isDebugEnabled()) { + logger.debug( + "Handling return value " + type + " with media type: " + mediaType); + } + ServletServerHttpRequest request = new ServletServerHttpRequest( + webRequest.getNativeRequest(HttpServletRequest.class)); + delegate.handleReturnValue( + getEmitter(timeout, flux, mediaType, request.getHeaders()), returnType, + mavContainer, webRequest); + } + + private void addHeaders(NativeWebRequest webRequest, Message message) { + HttpServletResponse response = webRequest + .getNativeResponse(HttpServletResponse.class); + ServletServerHttpRequest request = new ServletServerHttpRequest( + webRequest.getNativeRequest(HttpServletRequest.class)); + HttpHeaders headers = HeaderUtils.fromMessage(message.getHeaders(), + request.getHeaders()); + for (String name : headers.keySet()) { + for (Object object : headers.get(name)) { + response.addHeader(name, object.toString()); + } + } + } + + private boolean isInputSingle(NativeWebRequest webRequest, Object handler) { + Boolean single = (Boolean) webRequest.getAttribute( + WebRequestConstants.INPUT_SINGLE, NativeWebRequest.SCOPE_REQUEST); + if (single == null) { + return handler instanceof Supplier; + } + return single; + } + + private boolean isOutputSingle(Object handler) { + Class type = inspector.getOutputType(handler); + Class wrapper = inspector.getOutputWrapper(handler); + if (Stream.class.isAssignableFrom(type)) { + return false; + } + if (wrapper == type) { + return true; + } + if (Mono.class.equals(wrapper) || Optional.class.equals(wrapper)) { + return true; + } + return false; + } + + private MediaType findMediaType(NativeWebRequest webRequest) { + List accepts = Arrays.asList(MediaType.ALL); + MediaType mediaType = null; + if (webRequest.getHeader("Accept") != null) { + accepts = MediaType.parseMediaTypes(webRequest.getHeader("Accept")); + for (MediaType accept : accepts) { + if (!MediaType.ALL.equals(accept) + && MediaType.APPLICATION_JSON.isCompatibleWith(accept)) { + mediaType = MediaType.APPLICATION_JSON; + // Prefer JSON if that is acceptable + break; + } + else if (mediaType == null) { + mediaType = accept; + } + } + } + if (mediaType == null) { + mediaType = MediaType.APPLICATION_JSON; + } + return mediaType; + } + + private boolean isPlainText(NativeWebRequest webRequest) { + String value = webRequest.getHeader("Content-Type"); + if (value != null) { + return MediaType.valueOf(value).isCompatibleWith(MediaType.TEXT_PLAIN); + } + return false; + } + + private ResponseBodyEmitter getEmitter(Long timeout, Publisher flux, + MediaType mediaType, HttpHeaders request) { + Publisher exported = flux instanceof Mono ? Mono.from(flux) + : Flux.from(flux).timeout(Duration.ofMillis(timeout), Flux.empty()); + if (!MediaType.ALL.equals(mediaType) + && EVENT_STREAM.isCompatibleWith(mediaType)) { + // TODO: more subtle content negotiation + return new FluxResponseSseEmitter(request, MediaType.APPLICATION_JSON, + exported); + } + return new FluxResponseBodyEmitter(request, mediaType, exported); + } + +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/ResponseBodyEmitterSubscriber.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/ResponseBodyEmitterSubscriber.java new file mode 100644 index 000000000..3c5961561 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/ResponseBodyEmitterSubscriber.java @@ -0,0 +1,212 @@ +/* + * Copyright 2013-2016 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.function.web.flux.response; + +import java.io.IOException; +import java.util.concurrent.TimeoutException; + +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +import org.springframework.cloud.function.web.util.HeaderUtils; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.server.ServerHttpResponse; +import org.springframework.messaging.Message; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Subscriber that emits any value produced by the {@link Flux} into the delegated + * {@link ResponseBodyEmitter}. + * + * @author Dave Syer + */ +class ResponseBodyEmitterSubscriber implements Subscriber { + + private final MediaType mediaType; + + private Subscription subscription; + + private final ResponseBodyEmitter responseBodyEmitter; + + private boolean completed; + + private boolean firstElementWritten; + + private boolean single; + + private final boolean json; + + private Message first; + + private final HttpHeaders request; + + public ResponseBodyEmitterSubscriber(HttpHeaders request, MediaType mediaType, + Publisher observable, ResponseBodyEmitter responseBodyEmitter, + boolean json) { + + this.request = request; + this.mediaType = mediaType; + this.responseBodyEmitter = responseBodyEmitter; + this.json = json; + this.responseBodyEmitter.onTimeout(new Timeout()); + this.responseBodyEmitter.onCompletion(new Complete()); + this.single = observable instanceof Mono; + observable.subscribe(this); + } + + public void extendResponse(ServerHttpResponse response) { + headers(response); + } + + @Override + public void onSubscribe(Subscription subscription) { + this.subscription = subscription; + subscription.request(Long.MAX_VALUE); + } + + @Override + public void onNext(Object value) { + + Object object = value; + + if (object instanceof Message) { + Message message = (Message) object; + object = message.getPayload(); + this.first = message; + } + + try { + if (isJson()) { + if (!this.firstElementWritten) { + if (!single) { + responseBodyEmitter.send("["); + this.firstElementWritten = true; + } + } + else { + responseBodyEmitter.send(","); + } + if (!single && object.getClass() == String.class + && !((String) object).contains("\"")) { + object = "\"" + object + "\""; + } + } + if (!completed) { + responseBodyEmitter.send(object, mediaType); + } + } + catch ( + + IOException e) { + throw new RuntimeException(e.getMessage(), e); + } + } + + private void headers(ServerHttpResponse response) { + if (this.first != null) { + Message message = first; + try { + HttpHeaders headers = HeaderUtils.fromMessage(message.getHeaders(), + request); + for (String name : headers.keySet()) { + for (String value : headers.get(name)) { + response.getHeaders().add(name, value); + } + } + } + catch (Exception e) { + // Headers could not be set + } + } + } + + @Override + public void onError(Throwable e) { + if (!completed) { + completed = true; + try { + if (isJson()) { + if (!single) { + if (!this.firstElementWritten) { + responseBodyEmitter.send("[]"); + } + else { + responseBodyEmitter.send("]"); + } + } + } + if (e instanceof TimeoutException) { + responseBodyEmitter.complete(); + } + else { + responseBodyEmitter.completeWithError(e); + } + } + catch (IOException ex) { + throw new RuntimeException(ex.getMessage(), ex); + } + } + } + + @Override + public void onComplete() { + if (!completed) { + completed = true; + try { + if (isJson()) { + if (!single) { + if (!this.firstElementWritten) { + responseBodyEmitter.send("["); + } + responseBodyEmitter.send("]"); + } + } + } + catch (IOException e) { + throw new RuntimeException(e.getMessage(), e); + } + responseBodyEmitter.complete(); + } + } + + private boolean isJson() { + return json; + } + + class Complete implements Runnable { + + @Override + public void run() { + ResponseBodyEmitterSubscriber.this.subscription.cancel(); + } + } + + class Timeout implements Runnable { + + @Override + public void run() { + onComplete(); + ResponseBodyEmitterSubscriber.this.subscription.cancel(); + } + } + +} diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/HeaderUtils.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/HeaderUtils.java new file mode 100644 index 000000000..932859e33 --- /dev/null +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/HeaderUtils.java @@ -0,0 +1,84 @@ +/* + * Copyright 2016-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.function.web.util; + +import java.util.Arrays; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.springframework.http.HttpHeaders; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.ObjectUtils; + +/** + * @author Dave Syer + * + */ +public class HeaderUtils { + + public static HttpHeaders fromMessage(MessageHeaders headers, HttpHeaders request) { + HttpHeaders result = new HttpHeaders(); + for (String name : headers.keySet()) { + Object value = headers.get(name); + name = name.toLowerCase(); + if (MessageHeaders.ID.equals(name)) { + continue; + } + if (request.containsKey(name)) { + if (name.startsWith("x-")) { + if (!name.startsWith("x-forwarded")) { + Collection values = multi(value); + for (Object object : values) { + result.set(name, object.toString()); + } + } + } + } + else { + Collection values = multi(value); + for (Object object : values) { + result.set(name, object.toString()); + } + } + } + return result; + } + + private static Collection multi(Object value) { + if (value instanceof Collection) { + Collection collection = (Collection) value; + return collection; + } + else if (ObjectUtils.isArray(value)) { + Object[] values = ObjectUtils.toObjectArray(value); + return Arrays.asList(values); + } + return Arrays.asList(value); + } + + public static MessageHeaders fromHttp(HttpHeaders headers) { + Map map = new LinkedHashMap<>(); + for (String name : headers.keySet()) { + Collection values = multi(headers.get(name)); + name = name.toLowerCase(); + Object value = values == null ? null + : (values.size() == 1 ? values.iterator().next() : values); + map.put(name, value); + } + return new MessageHeaders(map); + } +} diff --git a/spring-cloud-function-web/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-cloud-function-web/src/main/resources/META-INF/additional-spring-configuration-metadata.json new file mode 100644 index 000000000..5c6dc17e8 --- /dev/null +++ b/spring-cloud-function-web/src/main/resources/META-INF/additional-spring-configuration-metadata.json @@ -0,0 +1,9 @@ +{"properties": [ + { + "name": "spring.cloud.function.web.path", + "type": "java.lang.String", + "description": "Path to web resources for functions (should start with / if not empty).", + "defaultValue": "" + }] +} + diff --git a/spring-cloud-function-web/src/main/resources/META-INF/spring.factories b/spring-cloud-function-web/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..812023313 --- /dev/null +++ b/spring-cloud-function-web/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +org.springframework.cloud.function.web.flux.ReactorAutoConfiguration \ No newline at end of file diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/mvc/MvcRestApplicationTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/mvc/MvcRestApplicationTests.java new file mode 100644 index 000000000..c7b6075d1 --- /dev/null +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/mvc/MvcRestApplicationTests.java @@ -0,0 +1,392 @@ +/* + * Copyright 2016-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.function.mvc; + +import java.net.URI; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.embedded.LocalServerPort; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.cloud.function.mvc.MvcRestApplicationTests.TestConfiguration; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.StringUtils; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.ResponseStatus; +import org.springframework.web.bind.annotation.RestController; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Tests for vanilla MVC handling (no function layer). Validates the MVC customizations + * that are added in this project independently of the specific concerns of function. + * + * @author Dave Syer + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(classes = TestConfiguration.class, webEnvironment = WebEnvironment.RANDOM_PORT) +public class MvcRestApplicationTests { + + private static final MediaType EVENT_STREAM = MediaType.valueOf("text/event-stream"); + @LocalServerPort + private int port; + @Autowired + private TestRestTemplate rest; + @Autowired + private TestConfiguration test; + + @Before + public void init() { + test.list.clear(); + } + + @Test + public void wordsSSE() throws Exception { + assertThat(rest.exchange( + RequestEntity.get(new URI("/words")).accept(EVENT_STREAM).build(), + String.class).getBody()).isEqualTo(sse("foo", "bar")); + } + + @Test + public void wordsJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/words")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + @Ignore("Fix error handling") + public void errorJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/bang")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[\"foo\"]"); + } + + @Test + public void words() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/words")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + public void foos() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/foos")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"); + } + + @Test + public void getMore() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/get/more")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + @Ignore("Should this even work? Or do we need to be explicit about the JSON?") + public void updates() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.post(new URI("/updates")).body("one\ntwo"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()).isEqualTo("onetwo"); + } + + @Test + public void updatesJson() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/updates")).contentType(MediaType.APPLICATION_JSON) + .body("[\"one\",\"two\"]"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()).isEqualTo("[\"one\",\"two\"]"); + } + + @Test + public void addFoos() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/addFoos")).contentType(MediaType.APPLICATION_JSON) + .body("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"); + } + + @Test + public void timeout() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/timeout")).build(), String.class) + .getBody()).isEqualTo("[\"foo\"]"); + } + + @Test + public void emptyJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/empty")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[]"); + } + + @Test + public void sentences() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/sentences")).build(), String.class) + .getBody()).isEqualTo("[[\"go\",\"home\"],[\"come\",\"back\"]]"); + } + + @Test + public void sentencesAcceptAny() throws Exception { + assertThat(rest.exchange( + RequestEntity.get(new URI("/sentences")).accept(MediaType.ALL).build(), + String.class).getBody()) + .isEqualTo("[[\"go\",\"home\"],[\"come\",\"back\"]]"); + } + + @Test + public void sentencesAcceptJson() throws Exception { + ResponseEntity result = rest + .exchange( + RequestEntity.get(new URI("/sentences")) + .accept(MediaType.APPLICATION_JSON).build(), + String.class); + assertThat(result.getBody()).isEqualTo("[[\"go\",\"home\"],[\"come\",\"back\"]]"); + assertThat(result.getHeaders().getContentType()) + .isGreaterThanOrEqualTo(MediaType.APPLICATION_JSON); + } + + @Test + public void uppercase() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/uppercase")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"[FOO]\",\"[BAR]\"]"); + } + + @Test + public void uppercaseFoos() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/upFoos")).contentType(MediaType.APPLICATION_JSON) + .body("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"), String.class); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"FOO\"},{\"value\":\"BAR\"}]"); + } + + @Test + public void transform() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/transform")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"[FOO]\",\"[BAR]\"]"); + } + + @Test + public void postMore() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/post/more")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"[FOO]\",\"[BAR]\"]"); + } + + @Test + public void uppercaseGet() { + assertThat(rest.getForObject("/uppercase/foo", String.class)).isEqualTo("[FOO]"); + } + + @Test + public void convertGet() { + assertThat(rest.getForObject("/wrap/123", String.class)).isEqualTo("..123.."); + } + + @Test + public void convertGetJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/entity/321")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("{\"value\":321}"); + } + + @Test + public void uppercaseJsonStream() throws Exception { + assertThat(rest + .exchange(RequestEntity.post(new URI("/maps")) + .contentType(MediaType.APPLICATION_JSON) + .body("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"), String.class) + .getBody()).isEqualTo("[{\"value\":\"FOO\"},{\"value\":\"BAR\"}]"); + } + + @Test + public void uppercaseSSE() throws Exception { + assertThat(rest.exchange(RequestEntity.post(new URI("/uppercase")) + .accept(EVENT_STREAM).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class).getBody()) + .isEqualTo(sse("[FOO]", "[BAR]")); + } + + private String sse(String... values) { + return "data:" + StringUtils.arrayToDelimitedString(values, "\n\ndata:") + "\n\n"; + } + + @EnableAutoConfiguration + @RestController + @Configuration + public static class TestConfiguration { + + private List list = new ArrayList<>(); + + @PostMapping({ "/uppercase", "/transform", "/post/more" }) + public Flux uppercase(@RequestBody List flux) { + return Flux.fromIterable(flux).log() + .map(value -> "[" + value.trim().toUpperCase() + "]"); + } + + @PostMapping("/upFoos") + public Flux upFoos(@RequestBody List list) { + return Flux.fromIterable(list).log() + .map(value -> new Foo(value.getValue().trim().toUpperCase())); + } + + @GetMapping("/uppercase/{id}") + public Mono uppercaseGet(@PathVariable String id) { + return Mono.just(id).map(value -> "[" + value.trim().toUpperCase() + "]"); + } + + @GetMapping("/wrap/{id}") + public Mono wrapGet(@PathVariable int id) { + return Mono.just(id).log().map(value -> ".." + value + ".."); + } + + @GetMapping("/entity/{id}") + public Mono> entity(@PathVariable Integer id) { + return Mono.just(id).log() + .map(value -> Collections.singletonMap("value", value)); + } + + @PostMapping("/maps") + public Flux> maps( + @RequestBody List> flux) { + return Flux.fromIterable(flux).map(value -> { + value.put("value", value.get("value").trim().toUpperCase()); + return value; + }); + } + + @GetMapping({ "/words", "/get/more" }) + public Flux words() { + return Flux.fromArray(new String[] { "foo", "bar" }); + } + + @GetMapping("/foos") + public Flux foos() { + return Flux.just(new Foo("foo"), new Foo("bar")); + } + + @PostMapping("/updates") + @ResponseStatus(HttpStatus.ACCEPTED) + public Flux updates(@RequestBody List list) { + Flux flux = Flux.fromIterable(list).cache(); + flux.subscribe(value -> this.list.add(value)); + return flux; + } + + @PostMapping("/addFoos") + @ResponseStatus(HttpStatus.ACCEPTED) + public Flux addFoos(@RequestBody List list) { + Flux flux = Flux.fromIterable(list).cache(); + flux.subscribe(value -> this.list.add(value.getValue())); + return flux; + } + + @GetMapping("/bang") + public Flux bang() { + return Flux.fromArray(new String[] { "foo", "bar" }).map(value -> { + if (value.equals("bar")) { + throw new RuntimeException("Bar"); + } + return value; + }); + } + + @GetMapping("/empty") + public Flux empty() { + return Flux.fromIterable(Collections.emptyList()); + } + + @GetMapping("/timeout") + public Flux timeout() { + return Flux.defer(() -> Flux.create(emitter -> { + emitter.next("foo"); + }).timeout(Duration.ofMillis(100L), Flux.empty())); + } + + @GetMapping("/sentences") + public Flux> sentences() { + return Flux.just(Arrays.asList("go", "home"), Arrays.asList("come", "back")); + } + + } + + public static class Foo { + private String value; + + public Foo(String value) { + this.value = value; + } + + Foo() { + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + } +} diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/scan/ComponentTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/scan/ComponentTests.java new file mode 100644 index 000000000..e05833d47 --- /dev/null +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/scan/ComponentTests.java @@ -0,0 +1,87 @@ +/* + * Copyright 2012-2015 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.function.scan; + +import java.net.URI; +import java.util.function.Function; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.embedded.LocalServerPort; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Component; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT) +public class ComponentTests { + + @LocalServerPort + private int port; + @Autowired + private Greeter greeter; + @Autowired + private TestRestTemplate rest; + + @Test + public void contextLoads() throws Exception { + assertThat(greeter).isNotNull(); + } + + @Test + public void greeter() throws Exception { + ResponseEntity result = rest + .exchange( + RequestEntity.post(new URI("/greeter")) + .contentType(MediaType.TEXT_PLAIN).body("World"), + String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("Hello World"); + } + + @SpringBootApplication + @ComponentScan + protected static class TestConfiguration { + } + + @Component("greeter") + protected static class Greeter implements Function, Flux> { + @Override + public Flux apply(Flux flux) { + return flux.map(name -> "Hello " + name); + } + } + +} diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PrefixTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PrefixTests.java new file mode 100644 index 000000000..606ca54fe --- /dev/null +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PrefixTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2012-2015 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.function.web; + +import java.net.URI; +import java.util.function.Supplier; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.embedded.LocalServerPort; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.http.HttpStatus; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT, properties = "spring.cloud.function.web.path=/functions") +public class PrefixTests { + + @LocalServerPort + private int port; + @Autowired + private TestRestTemplate rest; + + @Test + public void words() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.get(new URI("/functions/words")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + public void missing() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/words")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + } + + @EnableAutoConfiguration + @org.springframework.boot.test.context.TestConfiguration + protected static class TestConfiguration { + @Bean({ "words", "get/more" }) + public Supplier> words() { + return () -> Flux.fromArray(new String[] { "foo", "bar" }); + } + } +} diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java new file mode 100644 index 000000000..ff701b3d2 --- /dev/null +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java @@ -0,0 +1,599 @@ +/* + * Copyright 2016-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.function.web; + +import java.net.URI; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.embedded.LocalServerPort; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.StringUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT) +public class RestApplicationTests { + + private static final MediaType EVENT_STREAM = MediaType.TEXT_EVENT_STREAM; + @LocalServerPort + private int port; + @Autowired + private TestRestTemplate rest; + @Autowired + private ApplicationConfiguration test; + + @Before + public void init() { + test.list.clear(); + } + + @Test + public void staticResource() throws Exception { + assertThat(rest.getForObject("/test.html", String.class)).contains("Test"); + } + + @Test + public void wordsSSE() throws Exception { + assertThat(rest.exchange( + RequestEntity.get(new URI("/words")).accept(EVENT_STREAM).build(), + String.class).getBody()).isEqualTo(sse("foo", "bar")); + } + + @Test + public void wordsJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/words")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + @Ignore("Fix error handling") + public void errorJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/bang")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[\"foo\"]"); + } + + @Test + public void words() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/words")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + public void word() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/word")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("foo"); + } + + @Test + public void foos() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/foos")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"); + } + + @Test + public void qualifierFoos() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity.post(new URI("/foos")) + .contentType(MediaType.APPLICATION_JSON).body("[\"foo\",\"bar\"]"), + String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"[FOO]\"},{\"value\":\"[BAR]\"}]"); + } + + @Test + public void getMore() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/get/more")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + public void bareWords() throws Exception { + ResponseEntity result = rest + .exchange(RequestEntity.get(new URI("/bareWords")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @Test + @Ignore("Should this even work? Or do we need to be explicit about the JSON?") + public void updates() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.post(new URI("/updates")).body("one\ntwo"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()).isEqualTo("onetwo"); + } + + @Test + public void updatesJson() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/updates")).contentType(MediaType.APPLICATION_JSON) + .body("[\"one\",\"two\"]"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()).isEqualTo("[\"one\",\"two\"]"); + } + + @Test + public void addFoos() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/addFoos")).contentType(MediaType.APPLICATION_JSON) + .body("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"); + } + + @Test + public void bareUpdates() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/bareUpdates")).contentType(MediaType.APPLICATION_JSON) + .body("[\"one\",\"two\"]"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(test.list).hasSize(2); + assertThat(result.getBody()).isEqualTo("[\"one\",\"two\"]"); + } + + @Test + public void timeoutJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/timeout")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[\"foo\"]"); + } + + @Test + public void emptyJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/empty")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("[]"); + } + + @Test + public void sentences() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/sentences")).build(), String.class) + .getBody()).isEqualTo("[[\"go\",\"home\"],[\"come\",\"back\"]]"); + } + + @Test + public void sentencesAcceptAny() throws Exception { + assertThat(rest.exchange( + RequestEntity.get(new URI("/sentences")).accept(MediaType.ALL).build(), + String.class).getBody()) + .isEqualTo("[[\"go\",\"home\"],[\"come\",\"back\"]]"); + } + + @Test + public void sentencesAcceptJson() throws Exception { + ResponseEntity result = rest + .exchange( + RequestEntity.get(new URI("/sentences")) + .accept(MediaType.APPLICATION_JSON).build(), + String.class); + assertThat(result.getBody()).isEqualTo("[[\"go\",\"home\"],[\"come\",\"back\"]]"); + assertThat(result.getHeaders().getContentType()) + .isGreaterThanOrEqualTo(MediaType.APPLICATION_JSON); + } + + @Test + public void sentencesAcceptSse() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.get(new URI("/sentences")).accept(EVENT_STREAM).build(), + String.class); + assertThat(result.getBody()) + .isEqualTo(sse("[\"go\",\"home\"]", "[\"come\",\"back\"]")); + assertThat(result.getHeaders().getContentType().isCompatibleWith(EVENT_STREAM)) + .isTrue(); + } + + @Test + public void uppercase() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/uppercase")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]"); + } + + @Test + public void messages() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/messages")).contentType(MediaType.APPLICATION_JSON) + .header("x-foo", "bar").body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]"); + assertThat(result.getHeaders().getFirst("x-foo")).isEqualTo("bar"); + assertThat(result.getHeaders()).doesNotContainKey("id"); + } + + @Test + public void headers() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/headers")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]"); + assertThat(result.getHeaders().getFirst("foo")).isEqualTo("bar"); + assertThat(result.getHeaders()).doesNotContainKey("id"); + } + + @Test + public void uppercaseSingleValue() throws Exception { + ResponseEntity result = rest + .exchange( + RequestEntity.post(new URI("/uppercase")) + .contentType(MediaType.TEXT_PLAIN).body("foo"), + String.class); + assertThat(result.getBody()).isEqualTo("(FOO)"); + } + + @Test + @Ignore("WebFlux would split the request body into lines: TODO make this work the same") + public void uppercasePlainText() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.post(new URI("/uppercase")) + .contentType(MediaType.TEXT_PLAIN).body("foo\nbar"), + String.class); + assertThat(result.getBody()).isEqualTo("(FOO)(BAR)"); + } + + @Test + public void uppercaseFoos() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/upFoos")).contentType(MediaType.APPLICATION_JSON) + .body("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"), String.class); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"FOO\"},{\"value\":\"BAR\"}]"); + } + + @Test + public void uppercaseFoo() throws Exception { + // Single Foo can be parsed + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/upFoos")).contentType(MediaType.APPLICATION_JSON) + .body("{\"value\":\"foo\"}"), String.class); + assertThat(result.getBody()).isEqualTo("[{\"value\":\"FOO\"}]"); + } + + @Test + public void bareUppercaseFoos() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/bareUpFoos")).contentType(MediaType.APPLICATION_JSON) + .body("[{\"value\":\"foo\"},{\"value\":\"bar\"}]"), String.class); + assertThat(result.getBody()) + .isEqualTo("[{\"value\":\"FOO\"},{\"value\":\"BAR\"}]"); + } + + @Test + public void bareUppercaseFoo() throws Exception { + // Single Foo can be parsed and returns a single value if the function is defined + // that way + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/bareUpFoos")).contentType(MediaType.APPLICATION_JSON) + .body("{\"value\":\"foo\"}"), String.class); + assertThat(result.getBody()).isEqualTo("{\"value\":\"FOO\"}"); + } + + @Test + public void bareUppercase() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/bareUppercase")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]"); + } + + @Test + public void transform() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/transform")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]"); + } + + @Test + public void postMore() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity + .post(new URI("/post/more")).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class); + assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]"); + } + + @Test + public void postMoreFoo() { + assertThat(rest.getForObject("/post/more/foo", String.class)).isEqualTo("(FOO)"); + } + + @Test + public void uppercaseGet() { + assertThat(rest.getForObject("/uppercase/foo", String.class)).isEqualTo("(FOO)"); + } + + @Test + public void convertGet() { + assertThat(rest.getForObject("/wrap/123", String.class)).isEqualTo("..123.."); + } + + @Test + public void convertPost() throws Exception { + ResponseEntity result = rest.exchange(RequestEntity.post(new URI("/wrap")) + .contentType(MediaType.TEXT_PLAIN).body("123"), String.class); + assertThat(result.getBody()).isEqualTo("..123.."); + } + + @Test + public void convertPostJson() throws Exception { + // If you POST a single value to a Function,Flux> it can't + // determine if the output is single valued, so it has to send an array back + ResponseEntity result = rest + .exchange( + RequestEntity.post(new URI("/doubler")) + .contentType(MediaType.TEXT_PLAIN).body("123"), + String.class); + assertThat(result.getBody()).isEqualTo("[246]"); + } + + @Test + public void supplierFirst() { + assertThat(rest.getForObject("/not/a/function", String.class)) + .isEqualTo("[\"hello\"]"); + } + + @Test + public void convertGetJson() throws Exception { + assertThat(rest + .exchange(RequestEntity.get(new URI("/entity/321")) + .accept(MediaType.APPLICATION_JSON).build(), String.class) + .getBody()).isEqualTo("{\"value\":321}"); + } + + @Test + public void uppercaseJsonArray() throws Exception { + assertThat(rest.exchange( + RequestEntity.post(new URI("/maps")) + .contentType(MediaType.APPLICATION_JSON) + // The new line in the middle is optional + .body("[{\"value\":\"foo\"},\n{\"value\":\"bar\"}]"), + String.class).getBody()) + .isEqualTo("[{\"value\":\"FOO\"},{\"value\":\"BAR\"}]"); + } + + @Test + public void uppercaseSSE() throws Exception { + assertThat(rest.exchange(RequestEntity.post(new URI("/uppercase")) + .accept(EVENT_STREAM).contentType(MediaType.APPLICATION_JSON) + .body("[\"foo\",\"bar\"]"), String.class).getBody()) + .isEqualTo(sse("(FOO)", "(BAR)")); + } + + private String sse(String... values) { + return "data:" + StringUtils.arrayToDelimitedString(values, "\n\ndata:") + "\n\n"; + } + + @EnableAutoConfiguration + @TestConfiguration + public static class ApplicationConfiguration { + + private List list = new ArrayList<>(); + + @Bean({ "uppercase", "transform", "post/more" }) + public Function, Flux> uppercase() { + return flux -> flux.log() + .map(value -> "(" + value.trim().toUpperCase() + ")"); + } + + @Bean + public Function bareUppercase() { + return value -> "(" + value.trim().toUpperCase() + ")"; + } + + @Bean + public Function, Message> messages() { + return value -> MessageBuilder + .withPayload("(" + value.getPayload().trim().toUpperCase() + ")") + .copyHeaders(value.getHeaders()).build(); + } + + @Bean + public Function>, Flux>> headers() { + return flux -> flux.map(value -> MessageBuilder + .withPayload("(" + value.getPayload().trim().toUpperCase() + ")") + .setHeader("foo", "bar").build()); + } + + @Bean + public Function, Flux> upFoos() { + return flux -> flux.log() + .map(value -> new Foo(value.getValue().trim().toUpperCase())); + } + + @Bean + public Function bareUpFoos() { + return value -> new Foo(value.getValue().trim().toUpperCase()); + } + + @Bean + public Function, Flux> wrap() { + return flux -> flux.log().map(value -> ".." + value + ".."); + } + + @Bean + public Function, Flux> doubler() { + return flux -> flux.log().map(value -> 2 * value); + } + + @Bean + public Function, Flux>> entity() { + return flux -> flux.log() + .map(value -> Collections.singletonMap("value", value)); + } + + @Bean + public Function>, Flux>> maps() { + return flux -> flux.map(value -> { + value.put("value", value.get("value").trim().toUpperCase()); + return value; + }); + } + + @Bean({ "words", "get/more" }) + public Supplier> words() { + return () -> Flux.just("foo", "bar"); + } + + @Bean + public Supplier word() { + return () -> "foo"; + } + + @Bean + public Supplier> foos() { + return () -> Flux.just(new Foo("foo"), new Foo("bar")); + } + + @Bean + @Qualifier("foos") + public Function qualifier() { + return value -> new Foo("[" + value.trim().toUpperCase() + "]"); + } + + @Bean + public Supplier> bareWords() { + return () -> Arrays.asList("foo", "bar"); + } + + @Bean + public Consumer> updates() { + return flux -> flux.subscribe(value -> list.add(value)); + } + + @Bean + public Consumer> addFoos() { + return flux -> flux.subscribe(value -> list.add(value.getValue())); + } + + @Bean + public Consumer bareUpdates() { + return value -> list.add(value); + } + + @Bean + public Supplier> bang() { + return () -> Flux.fromArray(new String[] { "foo", "bar" }).map(value -> { + if (value.equals("bar")) { + throw new RuntimeException("Bar"); + } + return value; + }); + } + + @Bean + public Supplier> empty() { + return () -> Flux.fromIterable(Collections.emptyList()); + } + + @Bean("not/a/function") + public Supplier> supplier() { + return () -> Flux.just("hello"); + } + + @Bean("not/a") + public Function, Flux> function() { + return input -> Flux.just("bye"); + } + + @Bean + public Supplier> timeout() { + return () -> Flux.defer(() -> Flux.create(emitter -> { + emitter.next("foo"); + }).timeout(Duration.ofMillis(100L), Flux.empty())); + } + + @Bean + public Supplier>> sentences() { + return () -> Flux.just(Arrays.asList("go", "home"), + Arrays.asList("come", "back")); + } + + } + + public static class Foo { + private String value; + + public Foo(String value) { + this.value = value; + } + + Foo() { + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + } + +} diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/SingletonTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/SingletonTests.java new file mode 100644 index 000000000..f678390b4 --- /dev/null +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/SingletonTests.java @@ -0,0 +1,97 @@ +/* + * Copyright 2012-2015 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.function.web; + +import java.net.URI; +import java.util.function.Supplier; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.embedded.LocalServerPort; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.http.HttpStatus; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT) +public class SingletonTests { + + @LocalServerPort + private int port; + @Autowired + private TestRestTemplate rest; + + @Test + public void words() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.get(new URI("/words")).build(), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("[\"foo\",\"bar\"]"); + } + + @EnableAutoConfiguration + @org.springframework.boot.test.context.TestConfiguration + protected static class TestConfiguration { + @Bean + public static BeanDefinitionRegistryPostProcessor processor() { + return new BeanDefinitionRegistryPostProcessor() { + + @Override + public void postProcessBeanFactory( + ConfigurableListableBeanFactory beanFactory) + throws BeansException { + } + + @Override + public void postProcessBeanDefinitionRegistry( + BeanDefinitionRegistry registry) throws BeansException { + // Simulates what happens when you add a compiled function + RootBeanDefinition beanDefinition = new RootBeanDefinition(MySupplier.class); + registry.registerBeanDefinition("words", beanDefinition); + } + }; + } + } + + static class MySupplier implements Supplier> { + @Override + public Flux get() { + return Flux.just("foo", "bar"); + } + } +} diff --git a/spring-cloud-function-web/src/test/resources/static/test.html b/spring-cloud-function-web/src/test/resources/static/test.html new file mode 100644 index 000000000..9ca4440f5 --- /dev/null +++ b/spring-cloud-function-web/src/test/resources/static/test.html @@ -0,0 +1 @@ +Test \ No newline at end of file