diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java index ad457f439..c15706715 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java @@ -29,8 +29,9 @@ public interface FunctionCatalog { - default T lookupRaw(String name, MimeType... acceptedOutputTypes) { - return null; + default T lookup(String name, MimeType... acceptedOutputTypes) { + throw new UnsupportedOperationException( + "This instance of FunctionCatalog does not support this operation"); } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java index 535424efa..2c0add002 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java @@ -3,6 +3,7 @@ package org.springframework.cloud.function.context.catalog; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import org.apache.commons.logging.Log; @@ -11,10 +12,12 @@ import org.reactivestreams.Publisher; import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.core.convert.ConversionService; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; +import org.springframework.util.MimeType; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -90,34 +93,35 @@ class FunctionTypeConversionHelper { } @SuppressWarnings("rawtypes") - Object convertOutputIfNecessary(Object output) { + Object convertOutputIfNecessary(Object output, MimeType... acceptedOutputTypes) { List convertedResults = new ArrayList(); if (output instanceof Tuple2) { - convertedResults.add(this.doConvert(((Tuple2)output).getT1(), byte[].class, true)); - convertedResults.add(this.doConvert(((Tuple2)output).getT2(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple2)output).getT1(), acceptedOutputTypes[0])); + convertedResults.add(this.doConvert(((Tuple2)output).getT2(), acceptedOutputTypes[1])); } if (output instanceof Tuple3) { - convertedResults.add(this.doConvert(((Tuple3)output).getT3(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple3)output).getT3(), acceptedOutputTypes[2])); } if (output instanceof Tuple4) { - convertedResults.add(this.doConvert(((Tuple4)output).getT4(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple4)output).getT4(), acceptedOutputTypes[3])); } if (output instanceof Tuple5) { - convertedResults.add(this.doConvert(((Tuple5)output).getT5(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple5)output).getT5(), acceptedOutputTypes[4])); } if (output instanceof Tuple6) { - convertedResults.add(this.doConvert(((Tuple6)output).getT6(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple6)output).getT6(), acceptedOutputTypes[5])); } if (output instanceof Tuple7) { - convertedResults.add(this.doConvert(((Tuple7)output).getT7(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple7)output).getT7(), acceptedOutputTypes[6])); } if (output instanceof Tuple8) { - convertedResults.add(this.doConvert(((Tuple8)output).getT8(), byte[].class, true)); + convertedResults.add(this.doConvert(((Tuple8)output).getT8(), acceptedOutputTypes[7])); + } + + if (!CollectionUtils.isEmpty(convertedResults)) { + output = Tuples.fromArray(convertedResults.toArray()); } - output = CollectionUtils.isEmpty(convertedResults) - ? this.doConvert(output, byte[].class, true) - : Tuples.fromArray(convertedResults.toArray()); return output; } @@ -167,30 +171,43 @@ class FunctionTypeConversionHelper { return (Class) targetType; } - private Object doConvert(Object incoming, Type targetType) { - return this.doConvert(incoming, targetType, false); - } - @SuppressWarnings({ "unchecked", "rawtypes" }) - private Object doConvert(Object incoming, Type targetType, boolean toMessage) { + private Object doConvert(Object incoming, Type targetType) { Class actualType = this.getRawType(targetType); if (incoming instanceof Publisher) { if (!actualType.isAssignableFrom(Void.class)) { incoming = incoming instanceof Mono - ? Mono.from((Publisher) incoming).map(value -> this.doConvertArgument(value, targetType, actualType, toMessage)) - : Flux.from((Publisher) incoming).map(value -> this.doConvertArgument(value, targetType, actualType, toMessage)); + ? Mono.from((Publisher) incoming).map(value -> this.doConvertArgument(value, targetType, actualType)) + : Flux.from((Publisher) incoming).map(value -> this.doConvertArgument(value, targetType, actualType)); } } else { Assert.isTrue(!Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()), "Invoking reactive function as imperative is not allowed. Function name(s): " + this.functionRegistration.getNames()); - incoming = this.doConvertArgument(incoming, targetType, actualType, toMessage); + incoming = this.doConvertArgument(incoming, targetType, actualType); } return incoming; } - private Object doConvertArgument(Object incomingValue, Type targetType, Class actualInputType, boolean toMessage) { + @SuppressWarnings({ "unchecked", "rawtypes" }) + private Object doConvert(Object incoming, MimeType mimeType) { + MessageHeaders headers = new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, mimeType)); + if (incoming instanceof Publisher) { + incoming = incoming instanceof Mono + ? Mono.from((Publisher) incoming).map(value -> this.messageConverter.toMessage(value, headers)) + : Flux.from((Publisher) incoming).map(value -> this.messageConverter.toMessage(value, headers)); + } + else { + Assert.isTrue(!Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()), + "Invoking reactive function as imperative is not allowed. Function name(s): " + + this.functionRegistration.getNames()); + incoming = this.messageConverter.toMessage(incoming, headers); + } + return incoming; + } + + private Object doConvertArgument(Object incomingValue, Type targetType, Class actualInputType) { if (!Void.class.isAssignableFrom(actualInputType)) { if (incomingValue instanceof Message) { incomingValue = this.isMessage(targetType) @@ -204,9 +221,6 @@ class FunctionTypeConversionHelper { incomingValue = this.conversionService.convert(incomingValue, actualInputType); } } - if (toMessage) { - incomingValue = MessageBuilder.withPayload(incomingValue).build(); - } } else { incomingValue = null; diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java index 4f5a79678..9722790a1 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java @@ -42,6 +42,8 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.convert.ConversionService; import org.springframework.lang.Nullable; import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.util.MimeType; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; import reactor.core.publisher.Flux; @@ -76,7 +78,12 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector @SuppressWarnings("unchecked") @Override public T lookup(Class type, String definition) { - return (T) this.compose(type, definition, false); + return (T) this.compose(type, definition); + } + + @SuppressWarnings("unchecked") + public T lookup(String definition, MimeType... acceptedOutputTypes) { + return (T) this.compose(null, definition, acceptedOutputTypes); } @Override @@ -163,10 +170,10 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector // } @SuppressWarnings({ "unchecked", "rawtypes" }) - private Function compose(Class type, String definition, boolean raw) { + private Function compose(Class type, String definition, MimeType... acceptedOutputTypes) { Function resultFunction = null; if (this.registrationsByName.containsKey(definition)) { - resultFunction = new FunctionInvocationWrapper(this.registrationsByName.get(definition), false); + resultFunction = new FunctionInvocationWrapper(this.registrationsByName.get(definition), false, acceptedOutputTypes); } else { String[] names = StringUtils.delimitedListToStringArray(definition.replaceAll(",", "|").trim(), "|"); @@ -195,7 +202,7 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector FunctionRegistration registration = new FunctionRegistration<>(function, name).type(funcType); registrationsByFunction.putIfAbsent(function, registration); registrationsByName.putIfAbsent(name, registration); - function = new FunctionInvocationWrapper(registration, false); + function = new FunctionInvocationWrapper(registration, false, acceptedOutputTypes); if (resultFunction == null) { resultFunction = (Function) function; } @@ -216,7 +223,7 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector registration = new FunctionRegistration(resultFunction, composedNameBuilder.toString()).type(funcType); registrationsByFunction.putIfAbsent(resultFunction, registration); registrationsByName.putIfAbsent(composedNameBuilder.toString(), registration); - resultFunction = new FunctionInvocationWrapper(registration, true); + resultFunction = new FunctionInvocationWrapper(registration, true, acceptedOutputTypes); } previousFunctionType = funcType; prefix = "|"; @@ -247,10 +254,13 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector private final FunctionTypeConversionHelper functionTypeConversionHelper; - FunctionInvocationWrapper(FunctionRegistration functionRegistration, boolean composed) { + private final MimeType[] acceptedOutputTypes; + + FunctionInvocationWrapper(FunctionRegistration functionRegistration, boolean composed, MimeType... acceptedOutputTypes) { this.target = functionRegistration.getTarget(); this.functionRegistration = functionRegistration; this.composed = composed; + this.acceptedOutputTypes = acceptedOutputTypes; this.functionTypeConversionHelper = new FunctionTypeConversionHelper(this.functionRegistration, conversionService, messageConverter); } @@ -306,8 +316,9 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector } // ==== - //result = this.functionTypeConversionHelper.convertOutputIfNecessary(result); - // + if (!ObjectUtils.isEmpty(this.acceptedOutputTypes)) { + result = this.functionTypeConversionHelper.convertOutputIfNecessary(result, this.acceptedOutputTypes); + } return this.wrapOutputToReactiveIfNecessary(result); } diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java index 835848840..87d6a4ea6 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java +++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java @@ -231,7 +231,7 @@ public class LazyFunctionRegistryMultiInOutTests { public void testMultiToMultiWithMessageByteArrayPayload() { FunctionCatalog catalog = this.configureCatalog(); Function>, Flux>, Flux>>, Tuple2>, Mono>>> multiTuMulti = - catalog.lookupRaw("multiTuMulti", MimeTypeUtils.parseMimeType("foo/bar"), MimeTypeUtils.parseMimeType("bar/*")); + catalog.lookup("multiTuMulti", MimeTypeUtils.parseMimeType("application/json"), MimeTypeUtils.parseMimeType("application/json")); Flux> firstFlux = Flux.just( MessageBuilder.withPayload("Unlce".getBytes()).setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build(), @@ -249,7 +249,6 @@ public class LazyFunctionRegistryMultiInOutTests { MessageBuilder.withPayload(one.array()).setHeader(MessageHeaders.CONTENT_TYPE, "octet-stream/integer").build(), MessageBuilder.withPayload(two.array()).setHeader(MessageHeaders.CONTENT_TYPE, "octet-stream/integer").build()); - Tuple2>, Mono>> result = multiTuMulti.apply(Tuples.of(firstFlux, secondFlux, thirdFlux)); result.getT1().subscribe(v -> System.out.println("=> 1: " + v)); result.getT2().subscribe(v -> System.out.println("=> 2: " + v));