From d17f9e96ce2115e9de1c6b34f895c544c3b850e1 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Fri, 16 Apr 2021 09:39:08 +0200
Subject: [PATCH] Fix output conversion issue for cases where it needs to be
 defered

---
 .../catalog/SimpleFunctionRegistry.java       | 22 +++++++++----------
 1 file changed, 10 insertions(+), 12 deletions(-)

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 12d97af65..3052595cc 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
@@ -22,7 +22,6 @@ import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.lang.reflect.WildcardType;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -297,6 +296,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			String prefix = "";
 
 			Type originFunctionType = null;
+			boolean isReactive = false;
 			for (int i = 0; i < names.length; i++) {
 				String name = names[i];
 				Object function = this.locateFunction(name);
@@ -353,7 +353,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 				if (originFunctionType == null) {
 					originFunctionType = currentFunctionType;
 				}
-				if (name.equals(names[names.length - 1]) /*&& !names[0].equals("origin")*/) {
+
+				Type inputType = FunctionTypeUtils.getInputType(((FunctionInvocationWrapper) function).getFunctionType(), 0);
+				if (!isReactive) {
+					isReactive = FunctionTypeUtils.isReactive(inputType);
+				}
+				if (name.equals(names[names.length - 1]) && isReactive && !names[0].equals("origin")) {
 					((FunctionInvocationWrapper) function).setSkipOutputConversion(false);
 				}
 				else {
@@ -377,6 +382,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			((FunctionInvocationWrapper) resultFunction).acceptedOutputMimeTypes = acceptedOutputTypes;
 			FunctionRegistration<Object> registration = new FunctionRegistration<Object>(resultFunction, definition)
 				.type(originFunctionType);
+			if (((FunctionInvocationWrapper) resultFunction).composed || names.length == 1) {
+				((FunctionInvocationWrapper) resultFunction).setSkipOutputConversion(false);
+			}
 			registrationsByFunction.putIfAbsent(resultFunction, registration);
 		}
 		return resultFunction;
@@ -964,16 +972,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			return false;
 		}
 
-		private boolean isJson(Object value) {
-			String v = value instanceof byte[]
-					? new String((byte[]) value, StandardCharsets.UTF_8)
-							: (value instanceof String ? (String) value : null);
-			if (v != null && JsonMapper.isJsonString(v)) {
-				return true;
-			}
-			return false;
-		}
-
 		private boolean messageNeedsConversion(Type rawType, Type type, Message<?> message) {
 			Boolean skipConversion = message.getHeaders().containsKey(FunctionProperties.SKIP_CONVERSION_HEADER)
 				? message.getHeaders().get(FunctionProperties.SKIP_CONVERSION_HEADER, Boolean.class)