From 5e468fba73258d63a4ca85781da20d7b6f199abb Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Sat, 12 Dec 2020 13:48:47 +0100 Subject: [PATCH] Fix payload extraction during output conversion Also, restructured CloudEventsFunctionInvocationHelper's postProcessig logic --- .../CloudEventsFunctionInvocationHelper.java | 4 +-- .../context/catalog/FunctionTypeUtils.java | 12 ++++---- .../catalog/SimpleFunctionRegistry.java | 29 +++++++++++++++---- 3 files changed, 33 insertions(+), 12 deletions(-) diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java index 778c17216..94b38b814 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java @@ -52,8 +52,8 @@ class CloudEventsFunctionInvocationHelper implements FunctionInvocationHelper message) { - return message.getHeaders().containsKey(MessageUtils.MESSAGE_TYPE) - && message.getHeaders().get(MessageUtils.MESSAGE_TYPE).equals(CloudEventMessageUtils.CLOUDEVENT_VALUE); + return message.getHeaders().containsKey(MessageUtils.TARGET_PROTOCOL) || (message.getHeaders().containsKey(MessageUtils.MESSAGE_TYPE) + && message.getHeaders().get(MessageUtils.MESSAGE_TYPE).equals(CloudEventMessageUtils.CLOUDEVENT_VALUE)); } @Override diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java index f9edecb8c..cfdba9a3e 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java @@ -72,10 +72,12 @@ public final class FunctionTypeUtils { * @return 'true' if this type represents a {@link Collection}. Otherwise 'false'. */ public static boolean isTypeCollection(Type type) { + if (Collection.class.isAssignableFrom(getRawType(type))) { + return true; + } type = getGenericType(type); - Type rawType = type instanceof ParameterizedType ? ((ParameterizedType) type).getRawType() : type; - - return rawType instanceof Class && Collection.class.isAssignableFrom((Class) rawType); + Class rawType = type instanceof ParameterizedType ? getRawType(type) : (Class) type; + return Collection.class.isAssignableFrom(rawType); } /** @@ -298,8 +300,8 @@ public final class FunctionTypeUtils { } public static boolean isCollectionOfMessage(Type type) { - if (isMessage(type)) { - return isTypeCollection(type); + if (isMessage(type) && isTypeCollection(type)) { + return isMessage(getImmediateGenericType(type, 0)); } return false; } 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 e46900f51..7a66bd1db 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 @@ -880,6 +880,28 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect return convertedInput; } + private boolean isExtractPayload(Message message, Type type) { + if (FunctionTypeUtils.isCollectionOfMessage(type)) { + return true; + } + if (FunctionTypeUtils.isMessage(type)) { + return false; + } + + Object payload = message.getPayload(); + if (ObjectUtils.isArray(payload)) { + payload = CollectionUtils.arrayToList(payload); + } + if (payload instanceof Collection + && Message.class.isAssignableFrom(CollectionUtils.findCommonElementType((Collection) payload))) { + return true; + } + if (this.containsRetainMessageSignalInHeaders(message)) { + return false; + } + return true; + } + /** * This is an optional conversion which would only happen if `expected-content-type` is * set as a header in a message or explicitly provided as part of the lookup. @@ -888,11 +910,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect if (this.skipOutputConversion) { return output; } - if (output instanceof Message && !this.containsRetainMessageSignalInHeaders((Message) output)) { - if (!FunctionTypeUtils.isMessage(type) || - (FunctionTypeUtils.isMessage(type) && Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(type)))) { - output = ((Message) output).getPayload(); - } + if (output instanceof Message && isExtractPayload((Message) output, type)) { + output = ((Message) output).getPayload(); } if (!(output instanceof Publisher) && this.enhancer != null) { output = enhancer.apply(output);