Refactor FunctionCatalog implementation

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

@@ -32,6 +32,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
+import net.jodah.typetools.TypeResolver;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
@@ -41,6 +42,7 @@ import reactor.core.publisher.Mono;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
+import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.cloud.function.context.message.MessageUtils;
@@ -184,10 +186,12 @@ public class RequestProcessor {
 	private Mono<ResponseEntity<?>> response(FunctionWrapper request, Object handler,
 			Publisher<?> result, Boolean single, boolean getter) {
 		BodyBuilder builder = ResponseEntity.ok();
-		if (this.inspector.isMessage(handler)) {
+		if (((FunctionInvocationWrapper) handler).isInputTypeMessage()) {
 			result = Flux.from(result)
 					.map(message -> MessageUtils.unpack(handler, message))
-					.doOnNext(value -> addHeaders(builder, value))
+					.doOnNext(value -> {
+						addHeaders(builder, value);
+					})
 					.map(message -> message.getPayload());
 		}
 		else {
@@ -256,6 +260,7 @@ public class RequestProcessor {
 		}
 		else if (function instanceof FunctionInvocationWrapper) {
 			Publisher<?> result = (Publisher<?>) function.apply(flux);
+//			Publisher<?> result = null;
 			if (((FunctionInvocationWrapper) function).isConsumer()) {
 				if (result != null) {
 					((Mono) result).subscribe();
@@ -455,11 +460,33 @@ public class RequestProcessor {
 	}
 
 	private Type getItemType(Object function) {
-		Class<?> inputType = this.inspector.getInputType(function);
+
+		if (function == null || ((FunctionInvocationWrapper) function).getInputType() == Object.class) {
+			return Object.class;
+		}
+
+		Type itemType;
+		if (((FunctionInvocationWrapper) function).isInputTypePublisher() && ((FunctionInvocationWrapper) function).isInputTypeMessage()) {
+			itemType = FunctionTypeUtils.getImmediateGenericType(((FunctionInvocationWrapper) function).getInputType(), 0);
+			itemType = FunctionTypeUtils.getImmediateGenericType(itemType, 0);
+		}
+		else {
+			itemType = FunctionTypeUtils.getImmediateGenericType(((FunctionInvocationWrapper) function).getInputType(), 0);
+		}
+
+		if (itemType != null) {
+			return itemType;
+		}
+
+		Class<?> inputType = ((FunctionInvocationWrapper) function).isInputTypeMessage() || ((FunctionInvocationWrapper) function).isInputTypePublisher()
+				? TypeResolver.resolveRawClass(itemType, null)
+						: ((FunctionInvocationWrapper) function).getRawInputType();
 		if (!Collection.class.isAssignableFrom(inputType)) {
 			return inputType;
 		}
-		Type type = this.inspector.getRegistration(function).getType().getType();
+
+//		Type type = this.inspector.getRegistration(function).getType().getType();
+		Type type = ((FunctionInvocationWrapper) function).getInputType();
 		if (type instanceof ParameterizedType) {
 			type = ((ParameterizedType) type).getActualTypeArguments()[0];
 		}

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionController.java

@@ -63,7 +63,8 @@ public class FunctionController {
 	public Mono<ResponseEntity<?>> post(WebRequest request,
 			@RequestBody(required = false) String body) {
 		FunctionWrapper wrapper = wrapper(request);
-		return this.processor.post(wrapper, body, false);
+		Mono<ResponseEntity<?>> result = this.processor.post(wrapper, body, false);
+		return result;
 	}
 
 	@PostMapping(path = "/**", produces = MediaType.TEXT_EVENT_STREAM_VALUE)

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/source/FunctionExporterAutoConfiguration.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.function.web.source;
 
+import java.lang.reflect.Type;
 import java.util.function.Supplier;
 
 import reactor.core.publisher.Flux;
@@ -34,6 +35,7 @@ import org.springframework.cloud.function.web.source.FunctionExporterAutoConfigu
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.ResolvableType;
 import org.springframework.core.env.Environment;
 import org.springframework.web.reactive.function.client.WebClient;
 
@@ -66,9 +68,11 @@ public class FunctionExporterAutoConfiguration {
 	public FunctionRegistration<Supplier<Flux<?>>> origin(WebClient.Builder builder) {
 		HttpSupplier supplier = new HttpSupplier(builder.build(), this.props);
 		FunctionRegistration<Supplier<Flux<?>>> registration = new FunctionRegistration<>(supplier);
-		FunctionType type = FunctionType.supplier(this.props.getSource().getType()).wrap(Flux.class);
+		Type rawType = ResolvableType.forClassWithGenerics(Supplier.class, this.props.getSource().getType()).getType();
+//		FunctionType functionType = FunctionType.supplier(this.props.getSource().getType()).wrap(Flux.class);
+		FunctionType type = FunctionType.of(rawType);
 		if (this.props.getSource().isIncludeHeaders()) {
-			type = type.message();
+//			type = type.message();
 		}
 		registration = registration.type(type);
 		return registration;

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/source/SupplierExporter.java

@@ -168,7 +168,11 @@ public class SupplierExporter implements SmartLifecycle {
 	}
 
 	private Flux<ClientResponse> forward(Supplier<Publisher<Object>> supplier, String name) {
-		return Flux.from(supplier.get()).flatMap(value -> {
+		Flux o = (Flux) supplier.get();
+//		o.subscribe(v -> {
+//			System.out.println(v);
+//		});
+		return Flux.from(o).flatMap(value -> {
 			String destination = this.destinationResolver.destination(supplier, name, value);
 			if (this.debug) {
 				logger.info("Posting to: " + destination);