GH-1296: Fixed consumer with generics regression

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java

@@ -385,31 +385,33 @@ public static Type getInputType(Type functionType) {
 
 		ResolvableType resolvableFunctionType = ResolvableType.forType(functionType);
 
-		ResolvableType resolvableInputType = resolvableFunctionType.as(resolvableFunctionType.getRawClass());
-
-		if (resolvableInputType.getType() instanceof ParameterizedType) {
-			return resolvableInputType.getGeneric(0).getType();
+		if (FunctionTypeUtils.isFunction(functionType)) {
+			return extractInputType(resolvableFunctionType.as(Function.class).getGeneric(0));
 		}
-		else {
-			// will try another way. See GH-1251
-			if (FunctionTypeUtils.isFunction(functionType)) {
-				resolvableInputType = resolvableFunctionType.as(Function.class);
-			}
-			else {
-				if (KotlinDetector.isKotlinPresent() && Function1.class.isAssignableFrom(getRawType(functionType))) { // Kotlin
-					return ResolvableType.forType(getImmediateGenericType(functionType, 1)).getType();
-				}
-				else {
-					resolvableInputType = resolvableFunctionType.as(Consumer.class);
-				}
-			}
-			if (resolvableInputType.getType() instanceof ParameterizedType) {
-				return resolvableInputType.getGeneric(0).getType();
-			}
-			else  {
-				return Object.class;
-			}
+		if (FunctionTypeUtils.isConsumer(functionType)) {
+			return extractInputType(resolvableFunctionType.as(Consumer.class).getGeneric(0));
+		}
+		if (KotlinDetector.isKotlinPresent() && Function1.class.isAssignableFrom(getRawType(functionType))) { // Kotlin
+			return ResolvableType.forType(getImmediateGenericType(functionType, 1)).getType();
 		}
+
+		// no consumer or function
+		// might be one of the other supported types (as asserted in first line of method)
+		// for example FunctionRegistration or IntConsumer
+
+		// unclear what the contract is in such a case
+		// maybe returning null here might be "more" correct
+		return Object.class;
+	}
+
+	private static Type extractInputType(ResolvableType resolvableInputType) {
+		if (resolvableInputType.getType() instanceof TypeVariable) {
+			// In case the input type is a type variable (e.g. as in GH-1251) we need to resolve the type
+			// For the case that the type is unbound Object.class is used
+			return resolvableInputType.resolve(Object.class);
+		}
+
+		return resolvableInputType.getType();
 	}
 
 	@SuppressWarnings("rawtypes")

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtilsTests.java

@@ -46,6 +46,7 @@
 
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.messaging.Message;
+import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -63,6 +64,24 @@ public void testDiscoverFunctionalMethod() throws Exception {
 		assertThat(method.getName()).isEqualTo("accept");
 	}
 
+	private Type getBeanType(Class<?> configurationClass, String beanMethodName) {
+		return ReflectionUtils.findMethod(configurationClass, beanMethodName).getGenericReturnType();
+	}
+
+	@Test
+	public void testInputType() {
+		Type type;
+		Type inputType;
+
+		type = getBeanType(Gh1251Configuration.class, "consumer2");
+		inputType = FunctionTypeUtils.getInputType(type);
+		assertThat(inputType).isEqualTo(String.class);
+
+		type = getBeanType(Gh1251Configuration.class, "consumer3");
+		inputType = FunctionTypeUtils.getInputType(type);
+		assertThat(inputType).isEqualTo(String.class);
+	}
+
 	@Test
 	public void testFunctionTypeFrom() throws Exception {
 		Type type = FunctionTypeUtils.discoverFunctionTypeFromClass(SimpleConsumer.class);
@@ -288,6 +307,31 @@ public void accept(Flux<Message<String>> messageFlux) {
 		}
 	}
 
+	static class Gh1251Configuration {
+
+		Gh1251Consumer2<Integer> consumer2() {
+			return new Gh1251Consumer2<Integer>();
+		}
+
+		Gh1251Consumer3<String> consumer3() {
+			return new Gh1251Consumer3<>();
+		}
+	}
+
+	static class Gh1251Consumer2<E> implements Consumer<String> {
+
+		@Override
+		public void accept(String message) {
+		}
+	}
+
+	static class Gh1251Consumer3<E> implements Consumer<E> {
+
+		@Override
+		public void accept(E message) {
+		}
+	}
+
 	public interface ReactiveFunction<S, T> extends Function<Flux<S>, Flux<T>> {
 
 	}