diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index a070038ff..16cfd64e6 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java @@ -811,7 +811,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect } else if (!ObjectUtils.isEmpty(contentType)) { convertedOutput = messageConverter.toMessage(output, - new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeType.valueOf(contentType[0])))); + new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType[0]))); } return convertedOutput; diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java index d6eafc4e4..678611a15 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java @@ -103,9 +103,10 @@ public class ContextFunctionCatalogAutoConfiguration { // mcList.add(NegotiatingMessageConverterWrapper.wrap(new ByteArrayMessageConverter())); // mcList.add(NegotiatingMessageConverterWrapper.wrap(new StringMessageConverter())); + mcList.add(new StringMessageConverter()); mcList.add(new JsonMessageConverter(jsonMapper)); mcList.add(new ByteArrayMessageConverter()); - mcList.add(new StringMessageConverter()); + if (!CollectionUtils.isEmpty(mcList)) { messageConverter = new SmartCompositeMessageConverter(mcList); diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogInitializer.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogInitializer.java index 0f55aeebc..3159fb214 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogInitializer.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogInitializer.java @@ -172,10 +172,11 @@ public class ContextFunctionCatalogInitializer implements ApplicationContextInit // messageConverters.add(NegotiatingMessageConverterWrapper.wrap(new JsonMessageConverter(jsonMapper))); // messageConverters.add(NegotiatingMessageConverterWrapper.wrap(new ByteArrayMessageConverter())); // messageConverters.add(NegotiatingMessageConverterWrapper.wrap(new StringMessageConverter())); + messageConverters.add(new StringMessageConverter()); messageConverters.add(new JsonMessageConverter(jsonMapper)); messageConverters.add(new ByteArrayMessageConverter()); - messageConverters.add(new StringMessageConverter()); - CompositeMessageConverter messageConverter = new CompositeMessageConverter(messageConverters); + + SmartCompositeMessageConverter messageConverter = new SmartCompositeMessageConverter(messageConverters); ConversionService conversionService = new DefaultConversionService(); return new SimpleFunctionRegistry(conversionService, messageConverter, this.context.getBean(JsonMapper.class)); diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java index 91f9e8d5a..00288a85e 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java @@ -27,8 +27,8 @@ import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.support.MessageHeaderAccessor; -import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; +import org.springframework.util.StringUtils; /** * @@ -67,28 +67,36 @@ public class SmartCompositeMessageConverter extends CompositeMessageConverter { return null; } + @SuppressWarnings("unchecked") @Override @Nullable public Message toMessage(Object payload, @Nullable MessageHeaders headers) { for (MessageConverter converter : getConverters()) { - MessageHeaderAccessor accessor = new MessageHeaderAccessor(); - accessor.copyHeaders(headers); - if (this.isNotConcreteContentType(accessor, converter)) { - List supportedMimeTypes = ((AbstractMessageConverter) converter).getSupportedMimeTypes(); - for (MimeType supportedMimeType : supportedMimeTypes) { - accessor.setHeader(MessageHeaders.CONTENT_TYPE, supportedMimeType); - Message result = converter.toMessage(payload, accessor.getMessageHeaders()); + Object value = headers.get(MessageHeaders.CONTENT_TYPE).toString(); + String[] contentTypes = StringUtils.delimitedListToStringArray((String) value, ","); + for (String contentType : contentTypes) { + if (!MimeType.valueOf(contentType).isConcrete()) { + List supportedMimeTypes = ((AbstractMessageConverter) converter).getSupportedMimeTypes(); + for (MimeType supportedMimeType : supportedMimeTypes) { + MessageHeaderAccessor h = new MessageHeaderAccessor(); + h.copyHeaders(headers); + h.setHeader(MessageHeaders.CONTENT_TYPE, supportedMimeType); + Message result = converter.toMessage(payload, h.getMessageHeaders()); + if (result != null) { + return result; + } + } + } + else { + MessageHeaderAccessor h = new MessageHeaderAccessor(); + h.copyHeaders(headers); + h.setHeader(MessageHeaders.CONTENT_TYPE, contentType); + Message result = converter.toMessage(payload, h.getMessageHeaders()); if (result != null) { return result; } } } - else { - Message result = converter.toMessage(payload, headers); - if (result != null) { - return result; - } - } } return null; } @@ -96,32 +104,33 @@ public class SmartCompositeMessageConverter extends CompositeMessageConverter { @Override @Nullable public Message toMessage(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { - for (MessageConverter converter : getConverters()) { - MessageHeaderAccessor accessor = new MessageHeaderAccessor(); - accessor.copyHeaders(headers); - if (this.isNotConcreteContentType(accessor, converter)) { - List supportedMimeTypes = ((AbstractMessageConverter) converter).getSupportedMimeTypes(); - for (MimeType supportedMimeType : supportedMimeTypes) { - accessor.setHeader(MessageHeaders.CONTENT_TYPE, supportedMimeType); - Message result = ((AbstractMessageConverter) converter).toMessage(payload, accessor.getMessageHeaders(), conversionHint); + Object value = headers.get(MessageHeaders.CONTENT_TYPE).toString(); + String[] contentTypes = StringUtils.delimitedListToStringArray((String) value, ","); + for (String contentType : contentTypes) { + if (!MimeType.valueOf(contentType).isConcrete()) { + List supportedMimeTypes = ((AbstractMessageConverter) converter).getSupportedMimeTypes(); + for (MimeType supportedMimeType : supportedMimeTypes) { + MessageHeaderAccessor h = new MessageHeaderAccessor(); + h.copyHeaders(headers); + h.setHeader(MessageHeaders.CONTENT_TYPE, supportedMimeType); + Message result = ((SmartMessageConverter) converter).toMessage(payload, h.getMessageHeaders(), conversionHint); + if (result != null) { + return result; + } + } + } + else { + MessageHeaderAccessor h = new MessageHeaderAccessor(); + h.copyHeaders(headers); + h.setHeader(MessageHeaders.CONTENT_TYPE, contentType); + Message result = ((SmartMessageConverter) converter).toMessage(payload, h.getMessageHeaders(), conversionHint); if (result != null) { return result; } } } - else { - Message result = ((AbstractMessageConverter) converter).toMessage(payload, headers, conversionHint); - if (result != null) { - return result; - } - } } return null; } - - private boolean isNotConcreteContentType(MessageHeaderAccessor accessor, MessageConverter converter) { - return !accessor.getContentType().isConcrete() && converter instanceof AbstractMessageConverter - && !CollectionUtils.isEmpty(((AbstractMessageConverter) converter).getSupportedMimeTypes()); - } } diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java index a5ddf383a..92d7ffa71 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java @@ -48,6 +48,7 @@ import org.springframework.cloud.function.context.message.MessageUtils; import org.springframework.cloud.function.core.FluxConsumer; import org.springframework.cloud.function.core.FluxedConsumer; import org.springframework.cloud.function.json.JsonMapper; +import org.springframework.cloud.function.web.util.FunctionWebUtils; import org.springframework.cloud.function.web.util.HeaderUtils; import org.springframework.core.MethodParameter; import org.springframework.core.ReactiveAdapter; @@ -258,7 +259,9 @@ public class RequestProcessor { responseEntityMono = Mono.just(ResponseEntity.status(HttpStatus.ACCEPTED).build()); } else if (function instanceof FunctionInvocationWrapper) { - Publisher result = (Publisher) function.apply(flux); + + Publisher result = (Publisher) FunctionWebUtils.invokeFunction((FunctionInvocationWrapper) function, flux, + ((FunctionInvocationWrapper) function).isInputTypeMessage()); if (((FunctionInvocationWrapper) function).isConsumer()) { if (result != null) { ((Mono) result).subscribe(); 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 index 6e4d71602..33d1f5a1b 100644 --- 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 @@ -80,7 +80,7 @@ public class FunctionHandlerMapping extends RequestMappingHandlerMapping path = path.substring(this.prefix.length()); } Object function = FunctionWebUtils - .findFunction(request.getRequest().getMethod(), this.functions, request.getAttributes(), path); + .findFunction(request.getRequest().getMethod(), this.functions, request.getAttributes(), path, new String[] {}); if (function != null) { if (this.logger.isDebugEnabled()) { diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/function/FunctionEndpointInitializer.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/function/FunctionEndpointInitializer.java index 78d1c0a4f..1b9e6805a 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/function/FunctionEndpointInitializer.java +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/function/FunctionEndpointInitializer.java @@ -223,8 +223,9 @@ class FunctionEndpointFactory { function = this.functionCatalog.lookup(Function.class, handler); } else { + String[] accept = FunctionWebUtils.acceptContentTypes(request.headers().accept()); function = FunctionWebUtils.findFunction(request.method(), functionCatalog, - request.attributes(), request.path()); + request.attributes(), request.path(), accept); } return function; } @@ -252,7 +253,9 @@ class FunctionEndpointFactory { if (((FunctionInvocationWrapper) functionComponent).isSupplier()) { Supplier> supplier = (Supplier>) functionComponent; FunctionWrapper wrapper = RequestProcessor.wrapper(null, null, supplier); - Object result = wrapper.supplier().get(); + //Object result = wrapper.supplier().get(); + Object func = wrapper.supplier(); + Object result = FunctionWebUtils.invokeFunction((FunctionInvocationWrapper) func, null, ((FunctionInvocationWrapper) func).isInputTypeMessage()); if (!(result instanceof Publisher)) { result = Mono.just(result); } @@ -264,7 +267,9 @@ class FunctionEndpointFactory { wrapper.headers(request.headers().asHttpHeaders()); String argument = (String) request.attribute(WebRequestConstants.ARGUMENT).get(); wrapper.argument(Flux.just(argument)); - return ServerResponse.ok().body(wrapper.function().apply(wrapper.argument()), outputType); + Object func = wrapper.function(); + Object result = FunctionWebUtils.invokeFunction((FunctionInvocationWrapper) func, wrapper.argument(), ((FunctionInvocationWrapper) func).isInputTypeMessage()); + return ServerResponse.ok().body(result, outputType); } }); } diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionHandlerMapping.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionHandlerMapping.java index e7c44e797..29051d4b8 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionHandlerMapping.java +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionHandlerMapping.java @@ -92,7 +92,7 @@ public class FunctionHandlerMapping extends RequestMappingHandlerMapping } Object function = FunctionWebUtils.findFunction(HttpMethod.resolve(request.getMethod()), - this.functions, new HttpRequestAttributeDelegate(request), path); + this.functions, new HttpRequestAttributeDelegate(request), path, new String[] {}); if (function != null) { if (this.logger.isDebugEnabled()) { this.logger.debug("Found function for GET: " + path); @@ -110,6 +110,7 @@ public class FunctionHandlerMapping extends RequestMappingHandlerMapping this.request = request; } + @Override public Object put(String key, Object value) { this.request.setAttribute(key, value); return value; diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebUtils.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebUtils.java index 92401b306..6275cc296 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebUtils.java +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebUtils.java @@ -16,6 +16,7 @@ package org.springframework.cloud.function.web.util; +import java.util.List; import java.util.Map; import java.util.function.Function; import java.util.function.Supplier; @@ -23,8 +24,17 @@ import java.util.function.Supplier; import org.reactivestreams.Publisher; import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.web.constants.WebRequestConstants; import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; public final class FunctionWebUtils { @@ -34,20 +44,34 @@ public final class FunctionWebUtils { } public static Object findFunction(HttpMethod method, FunctionCatalog functionCatalog, - Map attributes, String path) { + Map attributes, String path, String[] acceptContentTypes) { if (method.equals(HttpMethod.GET) || method.equals(HttpMethod.POST)) { - return doFindFunction(method, functionCatalog, attributes, path); + return doFindFunction(method, functionCatalog, attributes, path, acceptContentTypes); } else { throw new IllegalStateException("HTTP method '" + method + "' is not supported;"); } } + public static String[] acceptContentTypes(List acceptHeaders) { + String[] acceptContentTypes = new String[] {}; + if (!CollectionUtils.isEmpty(acceptHeaders)) { + acceptContentTypes = acceptHeaders.stream().map(mediaType -> mediaType.toString()).toArray(String[]::new); + } + else { + acceptContentTypes = new String[] {MediaType.APPLICATION_JSON.toString()}; + } + + acceptContentTypes = new String[] {StringUtils.arrayToCommaDelimitedString(acceptContentTypes)}; +// return acceptContentTypes; + return new String[] {}; + } + private static Object doFindFunction(HttpMethod method, FunctionCatalog functionCatalog, - Map attributes, String path) { + Map attributes, String path, String[] acceptContentTypes) { path = path.startsWith("/") ? path.substring(1) : path; if (method.equals(HttpMethod.GET)) { - Supplier> supplier = functionCatalog.lookup(Supplier.class, path); + Supplier> supplier = functionCatalog.lookup(path, acceptContentTypes); if (supplier != null) { attributes.put(WebRequestConstants.SUPPLIER, supplier); return supplier; @@ -65,8 +89,7 @@ public final class FunctionWebUtils { name = builder.toString(); value = path.length() > name.length() ? path.substring(name.length() + 1) : null; - Function function = functionCatalog.lookup(Function.class, - name); + Function function = functionCatalog.lookup(name, acceptContentTypes); if (function != null) { attributes.put(WebRequestConstants.FUNCTION, function); if (value != null) { @@ -77,4 +100,32 @@ public final class FunctionWebUtils { } return null; } + + public static Object invokeFunction(FunctionInvocationWrapper function, Object input, boolean isMessage) { + Object result = function.apply(input); + return postProcessResult(result, isMessage); + } + + private static Object postProcessResult(Object result, boolean isMessage) { + if (result instanceof Flux) { + result = ((Flux) result).map(v -> postProcessResult(v, isMessage)); + } + else if (result instanceof Mono) { + result = ((Mono) result).map(v -> postProcessResult(v, isMessage)); + } + else if (result instanceof Message) { + if (!isMessage) { + result = ((Message) result).getPayload(); + } + else if (((Message) result).getPayload() instanceof byte[]) { + String str = new String((byte[]) ((Message) result).getPayload()); + result = MessageBuilder.withPayload(str).copyHeaders(((Message) result).getHeaders()).build(); + } + } + + if (result instanceof byte[]) { + result = new String((byte[]) result); + } + return result; + } }