SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix SmartCompositeMessageConverter to ensure it properly injects contentType

Browse Source 
Initial refactoring web - test pass
This commit is contained in:
Oleg Zhurakousky
2020-11-02 09:37:35 +01:00
parent a7104e489e
commit 63a1c3228d
9 changed files with 120 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java
View File

@@ -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();

2
spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionHandlerMapping.java
View File

@@ -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()) {

11
spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/function/FunctionEndpointInitializer.java
View File

@@ -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<? extends Flux<?>> supplier = (Supplier<Flux<?>>) 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);
}
});
}

3
spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionHandlerMapping.java
View File

@@ -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;

63
spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebUtils.java
View File

@@ -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<String, Object> attributes, String path) {
Map<String, Object> 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<MediaType> 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<String, Object> attributes, String path) {
Map<String, Object> attributes, String path, String[] acceptContentTypes) {
path = path.startsWith("/") ? path.substring(1) : path;
if (method.equals(HttpMethod.GET)) {
Supplier<Publisher<?>> supplier = functionCatalog.lookup(Supplier.class, path);
Supplier<Publisher<?>> 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<Object, Object> function = functionCatalog.lookup(Function.class,
name);
Function<Object, Object> 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;
}
}