Added initial support for lazy style FunctionCatalog/Registry which:

-  does not rely on any of the existing wrappers and instead relies on internal wrapper which performs  in-flight/just-in-time wrapping and unwrapping from reactive to imperative types
- performs transparent type conversion relying on MessageConverters and ConversionService
- supports multiple inputs/outputs

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

@@ -34,6 +34,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
@@ -178,7 +179,11 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
 			return Flux.empty();
 		}
 		if (this.supplier != null) {
-			return this.supplier.get();
+			Object result = this.supplier.get();
+			if (!(result instanceof Publisher)) {
+				result = Mono.just(result);
+			}
+			return (Publisher<?>) result;
 		}
 		throw new IllegalStateException("No function defined");
 	}

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

@@ -18,42 +18,62 @@ package org.springframework.cloud.function.context;
 
 import java.util.Set;
 
+import org.springframework.util.MimeType;
+
 /**
  * @author Dave Syer
  * @author Oleg Zhurakousky
  */
 public interface FunctionCatalog {
 
+
 	/**
-	 * Will look up the instance of the functional interface by name only.
-	 * @param <T> instance type
-	 * @param name the name of the functional interface. Must not be null;
+	 * Will look up the instance of the functional interface by name only and
+	 * acceptedOutputTypes.
+	 *
+	 * @param                     <T> instance type
+	 * @param functionDefinition  functionDefinition
+	 * @param acceptedOutputTypes acceptedOutputTypes
 	 * @return instance of the functional interface registered with this catalog
 	 */
-	default <T> T lookup(String name) {
-		return this.lookup(null, name);
+	default <T> T lookup(String functionDefinition, MimeType... acceptedOutputTypes) {
+		throw new UnsupportedOperationException("This instance of FunctionCatalog does not support this operation");
 	}
 
 	/**
-	 * Will look up the instance of the functional interface by name and type which can
-	 * only be Supplier, Consumer or Function. If type is not provided, the lookup will be
-	 * made based on name only.
-	 * @param <T> instance type
-	 * @param type the type of functional interface. Can be null
-	 * @param name the name of the functional interface. Must not be null;
+	 * Will look up the instance of the functional interface by name only.
+	 *
+	 * @param                    <T> instance type
+	 * @param functionDefinition the definition of the functional interface. Must
+	 *                           not be null;
 	 * @return instance of the functional interface registered with this catalog
 	 */
-	<T> T lookup(Class<?> type, String name);
+	default <T> T lookup(String functionDefinition) {
+		return this.lookup(null, functionDefinition);
+	}
+
+	/**
+	 * Will look up the instance of the functional interface by name and type which
+	 * can only be Supplier, Consumer or Function. If type is not provided, the
+	 * lookup will be made based on name only.
+	 *
+	 * @param                    <T> instance type
+	 * @param type               the type of functional interface. Can be null
+	 * @param functionDefinition the definition of the functional interface. Must
+	 *                           not be null;
+	 * @return instance of the functional interface registered with this catalog
+	 */
+	<T> T lookup(Class<?> type, String functionDefinition);
 
 	Set<String> getNames(Class<?> type);
 
 	/**
 	 * Return the count of functions registered in this catalog.
+	 *
 	 * @return the count of functions registered in this catalog
 	 */
 	default int size() {
-		throw new UnsupportedOperationException(
-				"This instance of FunctionCatalog does not support this operation");
+		throw new UnsupportedOperationException("This instance of FunctionCatalog does not support this operation");
 	}
 
 }

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

@@ -0,0 +1,320 @@
+/*
+ * Copyright 2016-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.context.catalog;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.lang.reflect.WildcardType;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuple3;
+import reactor.util.function.Tuple4;
+import reactor.util.function.Tuple5;
+import reactor.util.function.Tuple6;
+import reactor.util.function.Tuple7;
+import reactor.util.function.Tuple8;
+import reactor.util.function.Tuples;
+
+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 org.springframework.util.ObjectUtils;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+class FunctionTypeConversionHelper {
+
+	private static Log logger = LogFactory.getLog(FunctionTypeConversionHelper.class);
+
+	private final FunctionRegistration<?> functionRegistration;
+
+	private final Type[] functionArgumentTypes;
+
+	private final ConversionService conversionService;
+
+	private final MessageConverter messageConverter;
+
+	FunctionTypeConversionHelper(FunctionRegistration<?> functionRegistration, ConversionService conversionService,
+			MessageConverter messageConverter) {
+		this.conversionService = conversionService;
+		this.messageConverter = messageConverter;
+		this.functionRegistration = functionRegistration;
+		if ((this.functionRegistration.getType().getType()) instanceof ParameterizedType) {
+			this.functionArgumentTypes = ((ParameterizedType) this.functionRegistration.getType().getType())
+					.getActualTypeArguments();
+		}
+		else {
+			this.functionArgumentTypes = new Type[] { this.functionRegistration.getType().getInputType() };
+		}
+	}
+
+	@SuppressWarnings("rawtypes")
+	Object convertInputIfNecessary(Object input) {
+		List<Object> convertedResults = new ArrayList<Object>();
+		if (input instanceof Tuple2) {
+			convertedResults.add(this.doConvert(((Tuple2) input).getT1(), getInputArgumentType(0)));
+			convertedResults.add(this.doConvert(((Tuple2) input).getT2(), getInputArgumentType(1)));
+		}
+		if (input instanceof Tuple3) {
+			convertedResults.add(this.doConvert(((Tuple3) input).getT3(), getInputArgumentType(2)));
+		}
+		if (input instanceof Tuple4) {
+			convertedResults.add(this.doConvert(((Tuple4) input).getT4(), getInputArgumentType(3)));
+		}
+		if (input instanceof Tuple5) {
+			convertedResults.add(this.doConvert(((Tuple5) input).getT5(), getInputArgumentType(4)));
+		}
+		if (input instanceof Tuple6) {
+			convertedResults.add(this.doConvert(((Tuple6) input).getT6(), getInputArgumentType(5)));
+		}
+		if (input instanceof Tuple7) {
+			convertedResults.add(this.doConvert(((Tuple7) input).getT7(), getInputArgumentType(6)));
+		}
+		if (input instanceof Tuple8) {
+			convertedResults.add(this.doConvert(((Tuple8) input).getT8(), getInputArgumentType(7)));
+		}
+
+		input = CollectionUtils.isEmpty(convertedResults) ? this.doConvert(input, getInputArgumentType(0))
+				: Tuples.fromArray(convertedResults.toArray());
+		return input;
+	}
+
+	@SuppressWarnings("rawtypes")
+	Object convertOutputIfNecessary(Object output, MimeType... acceptedOutputTypes) {
+		if (ObjectUtils.isEmpty(acceptedOutputTypes)) {
+			return output;
+		}
+		List<Object> convertedResults = new ArrayList<Object>();
+		if (output instanceof Tuple2) {
+			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(), acceptedOutputTypes[2]));
+		}
+		if (output instanceof Tuple4) {
+			convertedResults.add(this.doConvert(((Tuple4) output).getT4(), acceptedOutputTypes[3]));
+		}
+		if (output instanceof Tuple5) {
+			convertedResults.add(this.doConvert(((Tuple5) output).getT5(), acceptedOutputTypes[4]));
+		}
+		if (output instanceof Tuple6) {
+			convertedResults.add(this.doConvert(((Tuple6) output).getT6(), acceptedOutputTypes[5]));
+		}
+		if (output instanceof Tuple7) {
+			convertedResults.add(this.doConvert(((Tuple7) output).getT7(), acceptedOutputTypes[6]));
+		}
+		if (output instanceof Tuple8) {
+			convertedResults.add(this.doConvert(((Tuple8) output).getT8(), acceptedOutputTypes[7]));
+		}
+
+		output = Tuples.fromArray(convertedResults.toArray());
+
+		return output;
+	}
+
+	int getInputArgumentCount() {
+		Type[] types = ((ParameterizedType) this.functionArgumentTypes[0]).getActualTypeArguments();
+		return types.length;
+	}
+
+	Object getInputArgument(int index) {
+		return 0;
+	}
+
+	Class<?> getInputArgumentRawType(int index) {
+		Type[] types = ((ParameterizedType) this.functionArgumentTypes[0]).getActualTypeArguments();
+		return (Class<?>) ((ParameterizedType) types[index]).getRawType();
+	}
+
+	Type getInputArgumentType(int index) {
+		if (this.functionArgumentTypes[0] instanceof ParameterizedType
+				&& (Publisher.class.isAssignableFrom((Class<?>) ((ParameterizedType) this.functionArgumentTypes[0]).getRawType())
+				|| ((ParameterizedType) this.functionArgumentTypes[0]).getTypeName().startsWith("reactor.util.function.Tuple"))) {
+			Type[] types = ((ParameterizedType) this.functionArgumentTypes[0]).getActualTypeArguments();
+
+			return (types[index]);
+		}
+		else {
+			return this.functionArgumentTypes[0];
+		}
+	}
+
+	Type getOutputArgumentType(int index) {
+		if (this.functionArgumentTypes[1] instanceof ParameterizedType) {
+			Type[] types = ((ParameterizedType) this.functionArgumentTypes[1]).getActualTypeArguments();
+
+			return (types[index]);
+		}
+		else {
+			return this.functionArgumentTypes[1];
+		}
+	}
+
+	private Class<?> getRawType(Type targetType) {
+		Class<?> rawType;
+		if (targetType instanceof ParameterizedType) {
+			if (Publisher.class.isAssignableFrom((Class<?>) ((ParameterizedType) targetType).getRawType())
+					|| Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) targetType).getRawType())) {
+
+				if (((ParameterizedType) targetType).getActualTypeArguments()[0] instanceof ParameterizedType) {
+					return this.getRawType(((ParameterizedType) targetType).getActualTypeArguments()[0]);
+				}
+				else {
+					rawType = (Class<?>) ((ParameterizedType) targetType).getActualTypeArguments()[0];
+				}
+			}
+			else {
+				rawType = (Class<?>) ((ParameterizedType) targetType).getRawType();
+			}
+		}
+		else {
+			if (targetType instanceof WildcardType) {
+				rawType = Object.class;
+			}
+			else {
+				rawType = (Class<?>) targetType;
+			}
+		}
+		return rawType;
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	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))
+								.doOnError(System.out::println)
+						: Flux.from((Publisher) incoming)
+								.map(value -> this.doConvertArgument(value, targetType, actualType))
+								.doOnError(System.out::println);
+
+			}
+		}
+		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);
+		}
+		return incoming;
+	}
+
+	@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)
+						? this.fromMessageToMessage((Message<?>) incomingValue, actualInputType)
+						: this.fromMessageToValue((Message<?>) incomingValue, actualInputType);
+			}
+			else {
+				if (!incomingValue.getClass().isAssignableFrom(actualInputType)) {
+					Assert.isTrue(this.conversionService.canConvert(incomingValue.getClass(), actualInputType),
+							"Failed to convert value of type " + incomingValue.getClass() + " to " + targetType);
+					incomingValue = this.conversionService.convert(incomingValue, actualInputType);
+				}
+			}
+		}
+		else {
+			incomingValue = null;
+		}
+		return incomingValue;
+	}
+
+	private boolean isMessage(Type targetType) {
+		if (targetType instanceof ParameterizedType) {
+			return Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) targetType).getRawType());
+		}
+		return false;
+	}
+
+	/*
+	 * Will conditionally convert Message's payload to a targetType unless such
+	 * payload is already of that type.
+	 */
+	private Object fromMessageToValue(Message<?> incomingMessage, Class<?> targetType) {
+		Object incomingValue = ((Message<?>) incomingMessage).getPayload();
+		if (!incomingValue.getClass().isAssignableFrom(targetType)) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Converting message '" + incomingMessage + "' with payload of type '"
+						+ incomingMessage.getPayload().getClass().getName() + "' to value of type '"
+						+ targetType.getName() + "' for invocation of " + functionRegistration.getNames());
+			}
+			if (incomingMessage.getPayload() instanceof Optional && !((Optional<?>) incomingMessage.getPayload()).isPresent()) {
+				incomingValue = incomingMessage;
+			}
+			else {
+				incomingValue = this.messageConverter.fromMessage((Message<?>) incomingMessage, targetType);
+			}
+		}
+		return incomingValue;
+	}
+
+	/*
+	 * Will conditionally convert Message's payload to a targetType unless such
+	 * payload is already of that type wrapping the result of conversion into a
+	 * Message with converted type as a payload.
+	 */
+	private Message<?> fromMessageToMessage(Message<?> incomingMessage, Class<?> targetType) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Converting message '" + incomingMessage + "' with payload of type '"
+					+ incomingMessage.getPayload().getClass().getName() + "' to message with payload of type '"
+					+ targetType.getName() + "' for invocation of " + functionRegistration.getNames());
+		}
+		return MessageBuilder.withPayload(this.fromMessageToValue(incomingMessage, targetType))
+				.copyHeaders(incomingMessage.getHeaders()).build();
+	}
+
+}

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

@@ -0,0 +1,476 @@
+/*
+ * Copyright 2019-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.context.catalog;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer;
+import org.springframework.cloud.function.context.FunctionRegistration;
+import org.springframework.cloud.function.context.FunctionRegistry;
+import org.springframework.cloud.function.context.FunctionType;
+import org.springframework.cloud.function.context.config.FunctionContextUtils;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.type.StandardMethodMetadata;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.StringUtils;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class LazyFunctionRegistry
+		implements FunctionRegistry, FunctionInspector, ApplicationContextAware, SmartInitializingSingleton {
+
+	private static Log logger = LogFactory.getLog(AbstractSpringFunctionAdapterInitializer.class);
+
+	private ConfigurableApplicationContext applicationContext;
+
+	private Map<Object, FunctionRegistration<Object>> registrationsByFunction = new HashMap<>();
+
+	private Map<String, FunctionRegistration<Object>> registrationsByName = new HashMap<>();
+
+	private final ConversionService conversionService;
+
+	private final CompositeMessageConverter messageConverter;
+
+	public LazyFunctionRegistry(ConversionService conversionService,
+			@Nullable CompositeMessageConverter messageConverter) {
+		this.conversionService = conversionService;
+		this.messageConverter = messageConverter;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> T lookup(Class<?> type, String definition) {
+		return (T) this.compose(type, definition);
+	}
+
+	@Override
+	public int size() {
+		return this.applicationContext.getBeanNamesForType(Supplier.class).length +
+				this.applicationContext.getBeanNamesForType(Function.class).length +
+				this.applicationContext.getBeanNamesForType(Consumer.class).length;
+	}
+
+	@SuppressWarnings("unchecked")
+	public <T> T lookup(String definition, MimeType... acceptedOutputTypes) {
+		Assert.notEmpty(acceptedOutputTypes, "'acceptedOutputTypes' must not be null or empty");
+		return (T) this.compose(null, definition, acceptedOutputTypes);
+	}
+
+	@Override
+	public boolean isMessage(Object function) {
+		if (function instanceof FunctionInvocationWrapper) {
+			function = ((FunctionInvocationWrapper) function).target;
+		}
+		return FunctionInspector.super.isMessage(function);
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Override
+	public void afterSingletonsInstantiated() {
+		Map<String, FunctionRegistration> beansOfType = this.applicationContext
+				.getBeansOfType(FunctionRegistration.class);
+		for (FunctionRegistration fr : beansOfType.values()) {
+			this.registrationsByFunction.putIfAbsent(fr.getTarget(), fr);
+			for (Object name : fr.getNames()) {
+				this.registrationsByName.putIfAbsent((String) name, fr);
+			}
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<String> getNames(Class<?> type) {
+		Set<String> registeredNames = registrationsByFunction.values().stream().flatMap(reg -> reg.getNames().stream())
+				.collect(Collectors.toSet());
+		registeredNames.addAll(CollectionUtils.arrayToList(this.applicationContext.getBeanNamesForType(type)));
+		return registeredNames;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> void register(FunctionRegistration<T> registration) {
+		this.registrationsByFunction.put(registration.getTarget(), (FunctionRegistration<Object>) registration);
+		for (String name : registration.getNames()) {
+			this.registrationsByName.put(name, (FunctionRegistration<Object>) registration);
+		}
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+	}
+
+	@Override
+	public FunctionRegistration<?> getRegistration(Object function) {
+		if (function instanceof FunctionInvocationWrapper) {
+			function = ((FunctionInvocationWrapper) function).target;
+		}
+		return this.registrationsByFunction.get(function);
+	}
+
+	private Collection<String> getAliases(String key) {
+		Collection<String> names = new LinkedHashSet<>();
+		String value = getQualifier(key);
+		if (value.equals(key) && this.applicationContext != null) {
+			names.addAll(Arrays.asList(this.applicationContext.getBeanFactory().getAliases(key)));
+		}
+		names.add(value);
+		return names;
+	}
+
+	private String getQualifier(String key) {
+		if (this.applicationContext != null && this.applicationContext.getBeanFactory().containsBeanDefinition(key)) {
+			BeanDefinition beanDefinition = this.applicationContext.getBeanFactory().getBeanDefinition(key);
+			Object source = beanDefinition.getSource();
+			if (source instanceof StandardMethodMetadata) {
+				StandardMethodMetadata metadata = (StandardMethodMetadata) source;
+				Qualifier qualifier = AnnotatedElementUtils.findMergedAnnotation(metadata.getIntrospectedMethod(),
+						Qualifier.class);
+				if (qualifier != null && qualifier.value().length() > 0) {
+					return qualifier.value();
+				}
+			}
+		}
+		return key;
+	}
+
+	private Object locateFunction(String name) {
+		Object function = null;
+		if (this.applicationContext.containsBean(name)) {
+			function = this.applicationContext.getBean(name);
+		}
+		if (function == null) {
+			function = this.registrationsByName.get(name);
+		}
+		return function;
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	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,
+					acceptedOutputTypes);
+		}
+		else {
+			if (StringUtils.isEmpty(definition)) {
+				String[] functionNames = this.applicationContext.getBeanNamesForType(Function.class);
+				Assert.notEmpty(functionNames, "Can't find any functions in BeanFactory");
+				Assert.isTrue(functionNames.length == 1, "Found more then one function in BeanFactory");
+				definition = functionNames[0];
+			}
+			String[] names = StringUtils.delimitedListToStringArray(definition.replaceAll(",", "|").trim(), "|");
+
+			FunctionType previousFunctionType = null;
+
+			StringBuilder composedNameBuilder = new StringBuilder();
+			String prefix = "";
+			for (String name : names) {
+				Object function = this.locateFunction(name);
+				if (function == null) {
+					return null;
+				}
+				composedNameBuilder.append(prefix);
+				composedNameBuilder.append(name);
+
+				FunctionRegistration<Object> registration;
+				FunctionType funcType;
+				if (function instanceof FunctionRegistration) {
+					registration = (FunctionRegistration<Object>) function;
+					funcType = registration.getType();
+					function = registration.getTarget();
+				}
+				else {
+					String[] aliasNames = this.getAliases(name).toArray(new String[] {});
+					funcType = beanDefinitionExists(aliasNames)
+							? FunctionType.of(FunctionContextUtils.findType(
+									(ConfigurableListableBeanFactory) applicationContext.getBeanFactory(), aliasNames))
+							: new FunctionType(function.getClass());
+					registration = new FunctionRegistration<>(function, name).type(funcType);
+				}
+
+				registrationsByFunction.putIfAbsent(function, registration);
+				registrationsByName.putIfAbsent(name, registration);
+				function = new FunctionInvocationWrapper(registration, false, acceptedOutputTypes);
+				if (resultFunction == null) {
+					resultFunction = (Function<?, ?>) function;
+				}
+				else {
+					resultFunction = resultFunction.andThen((Function) function);
+					if (this.getOutputWrapper(function).isAssignableFrom(Flux.class)) {
+						funcType = FunctionType.compose(previousFunctionType.wrap(Flux.class), funcType);
+						logger.info("Since composed function " + composedNameBuilder.toString()
+								+ " consists of at least one function "
+								+ "with return type Publisher, its resulting signature is Function<?, Publisher<?>>");
+					}
+					else if (this.getOutputWrapper(function).isAssignableFrom(Mono.class)) {
+						funcType = FunctionType.compose(previousFunctionType.wrap(Mono.class), funcType);
+					}
+					else {
+						funcType = FunctionType.compose(previousFunctionType, funcType);
+					}
+
+					registration = new FunctionRegistration<Object>(resultFunction, composedNameBuilder.toString())
+							.type(funcType);
+					registrationsByFunction.putIfAbsent(resultFunction, registration);
+					registrationsByName.putIfAbsent(composedNameBuilder.toString(), registration);
+					resultFunction = new FunctionInvocationWrapper(registration, true, acceptedOutputTypes);
+				}
+				previousFunctionType = funcType;
+				prefix = "|";
+			}
+		}
+
+		return resultFunction;
+	}
+
+	private boolean beanDefinitionExists(String... names) {
+		for (String name : names) {
+			if (this.applicationContext.getBeanFactory().containsBeanDefinition(name)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * Single wrapper for all Suppliers, Functions and Consumers managed by this
+	 * catalog.
+	 *
+	 * @author Oleg Zhurakousky
+	 *
+	 */
+	public class FunctionInvocationWrapper implements Function<Object, Object>, Consumer<Object>, Supplier<Object> {
+
+		private final Object target;
+
+		private final FunctionRegistration<?> functionRegistration;
+
+		private final boolean composed;
+
+		private final FunctionTypeConversionHelper functionTypeConversionHelper;
+
+		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);
+		}
+
+		@Override
+		public void accept(Object input) {
+			this.doApply(input, true);
+		}
+
+		@Override
+		public Object apply(Object input) {
+			return this.doApply(input, false);
+		}
+
+		@Override
+		public Object get() {
+			// wrap/unwrap to/from reactive
+			Object input = Mono.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())
+					? Mono.empty()
+					: (Flux.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()) ? Flux.empty()
+							: null);
+
+			return this.doApply(input, false);
+		}
+
+		public Object getTarget() {
+			return this.target;
+		}
+
+		@SuppressWarnings("unchecked")
+		private Object doApply(Object input, boolean consumer) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Applying function: " + this.functionRegistration.getNames());
+			}
+
+			if (input != null) {
+				input = this.wrapInputToReactiveIfNecessary(input);
+			}
+
+			Object result;
+			if (input instanceof Publisher) {
+				if (input != null && !this.composed) {
+					input = this.functionTypeConversionHelper.convertInputIfNecessary(input);
+				}
+				result = this.applyReactive((Publisher<Object>) input, consumer);
+			}
+			else {
+				if (Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())) {
+					throw new IllegalArgumentException("Invoking reactive function as imperative is not "
+							+ "allowed. Function name(s): " + this.functionRegistration.getNames());
+				}
+				else {
+					if (input != null && !this.composed) {
+						input = this.functionTypeConversionHelper.convertInputIfNecessary(input);
+					}
+					result = this.applyImperative(input, consumer);
+				}
+			}
+
+			result = this.functionTypeConversionHelper.convertOutputIfNecessary(result, this.acceptedOutputTypes);
+
+			if (!(result instanceof Publisher) && this.functionRegistration.getTarget() instanceof Supplier) {
+				/*
+				 * This is ONLY relevant for web, so consider exposing some property or may be
+				 * the fact that this is a rare case (Supplier) leave it temporarily as is.
+				 */
+
+				return Flux.just(this.wrapOutputToReactiveIfNecessary(result));
+			}
+			else {
+				return this.wrapOutputToReactiveIfNecessary(result);
+			}
+
+		}
+
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		private Object wrapOutputToReactiveIfNecessary(Object result) {
+			if (Flux.class.isAssignableFrom(this.functionRegistration.getType().getOutputWrapper())) {
+				result = result instanceof Publisher ? Flux.from((Publisher) result) : Flux.just(result);
+			}
+			else if (Mono.class.isAssignableFrom(this.functionRegistration.getType().getOutputWrapper())) {
+				result = result instanceof Publisher ? Mono.from((Publisher) result) : Mono.just(result);
+			}
+			return result;
+		}
+
+		/*
+		 * For functions of type `Function<?, Publisher<?>>` the input will be converted
+		 * to Publisher as well resulting in `Function<Publisher<?>, Publisher<?>>`
+		 */
+		private Object wrapInputToReactiveIfNecessary(Object input) {
+			if (input != null && !(input instanceof Publisher)) { // for Function<Object, Publisher>
+				if (Flux.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())) {
+					input = Flux.just(input);
+				}
+				else if (Mono.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())) {
+					input = Mono.just(input);
+				}
+			}
+			return input;
+		}
+
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		private Object applyImperative(Object input, boolean consumer) {
+			Object result = null;
+			if (this.target instanceof Function) {
+				if (Flux.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())) {
+					result = ((Function) this.target).apply(Flux.just(input));
+					// we may need to convert output as well
+				}
+				else {
+					result = ((Function) this.target).apply(input);
+				}
+			}
+			else if (this.target instanceof Consumer) {
+				((Consumer) this.target).accept(input);
+			}
+			else if (this.target instanceof Supplier) {
+				result = ((Supplier) this.target).get();
+			}
+			else {
+				throw new UnsupportedOperationException(
+						"Target of type " + this.target.getClass() + " is not supported");
+			}
+			return result;
+		}
+
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		private Object applyReactive(Publisher<Object> publisher, boolean consumer) {
+			Object result;
+			if (this.target instanceof Function) {
+				if (Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())) {
+					result = ((Function) this.target).apply(publisher);
+				}
+				else {
+					if (Void.class.isAssignableFrom(this.functionRegistration.getType().getInputType()) && !functionRegistration.getType().isMessage()) {
+						result = ((Function) this.target).apply(null);
+						result = publisher instanceof Mono ? Mono.just(result) : Flux.just(result);
+					}
+					else {
+						result = publisher instanceof Mono
+								? Mono.from(publisher).map(value -> ((Function) this.target).apply(value))
+								: Flux.from(publisher).map(value -> ((Function) this.target).apply(value));
+					}
+				}
+			}
+			else if (this.target instanceof Consumer) {
+				if (Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper())) {
+					((Consumer<Publisher<?>>) this.target).accept(publisher);
+					result = null;
+				}
+				else {
+					result = publisher instanceof Flux ? Flux.from(publisher).doOnNext((Consumer) this.target).then()
+							: Mono.from(publisher).doOnNext((Consumer) this.target).then();
+					if (consumer) {
+						((Mono<?>) result).subscribe();
+					}
+				}
+			}
+			else if (this.target instanceof Supplier) {
+				result = ((Supplier<?>) this.target).get();
+			}
+			else {
+				throw new UnsupportedOperationException(
+						"Target of type " + this.target.getClass() + " is not supported");
+			}
+			return result;
+		}
+	}
+}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java

@@ -23,6 +23,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Consumer;
@@ -54,6 +55,7 @@ import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.AbstractComposableFunctionRegistry;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.function.context.catalog.FunctionUnregistrationEvent;
+import org.springframework.cloud.function.context.catalog.LazyFunctionRegistry;
 import org.springframework.cloud.function.json.GsonMapper;
 import org.springframework.cloud.function.json.JacksonMapper;
 import org.springframework.context.ApplicationEventPublisher;
@@ -64,13 +66,17 @@ import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.FilterType;
 import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.core.type.StandardMethodMetadata;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.converter.ByteArrayMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MappingJackson2MessageConverter;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.converter.StringMessageConverter;
 
+
 /**
  * @author Dave Syer
  * @author Mark Fisher
@@ -84,11 +90,26 @@ public class ContextFunctionCatalogAutoConfiguration {
 
 	static final String PREFERRED_MAPPER_PROPERTY = "spring.http.converters.preferred-json-mapper";
 
-	@Bean
+//	@Bean
 	public FunctionRegistry functionCatalog() {
 		return new BeanFactoryFunctionCatalog();
 	}
 
+	@Bean
+	public FunctionRegistry functionCatalog(@Nullable ConversionService conversionService, @Nullable CompositeMessageConverter messageConverter,
+			Map<String, MessageConverter> additionalConverters) {
+		conversionService = conversionService == null ? new DefaultConversionService() : conversionService;
+		if (messageConverter == null) {
+			List<MessageConverter> messageConverters = new ArrayList<>();
+			messageConverters.addAll(additionalConverters.values());
+			messageConverters.add(new MappingJackson2MessageConverter());
+			messageConverters.add(new ByteArrayMessageConverter());
+			messageConverters.add(new StringMessageConverter());
+			messageConverter = new CompositeMessageConverter(messageConverters);
+		}
+		return new LazyFunctionRegistry(conversionService, messageConverter);
+	}
+
 	@Bean(RoutingFunction.FUNCTION_NAME)
 	@ConditionalOnProperty(name = "spring.cloud.function.routing.enabled", havingValue = "true")
 	RoutingFunction gateway(FunctionCatalog functionCatalog, FunctionInspector functionInspector) {

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FunctionContextUtils.java

@@ -42,8 +42,19 @@ import org.springframework.util.ReflectionUtils;
 public abstract class FunctionContextUtils {
 
 	public static Type findType(String name, ConfigurableListableBeanFactory registry) {
-		AbstractBeanDefinition definition = (AbstractBeanDefinition) registry
-				.getBeanDefinition(name);
+		return findType(registry, name);
+	}
+
+	public static Type findType(ConfigurableListableBeanFactory registry, String... names) {
+		AbstractBeanDefinition definition = null;
+		String actualName = null;
+		for (String name : names) {
+			if (registry.containsBeanDefinition(name)) {
+				definition = (AbstractBeanDefinition) registry
+						.getBeanDefinition(name);
+				actualName = name;
+			}
+		}
 
 		Object source = definition.getSource();
 
@@ -52,7 +63,7 @@ public abstract class FunctionContextUtils {
 			param = findBeanType(definition, ((MethodMetadata) source).getDeclaringClassName(), ((MethodMetadata) source).getMethodName());
 		}
 		else if (source instanceof Resource) {
-			param = registry.getType(name);
+			param = registry.getType(actualName);
 		}
 		else {
 			ResolvableType type = (ResolvableType) getField(definition, "targetType");
@@ -66,7 +77,7 @@ public abstract class FunctionContextUtils {
 					param = beanClass;
 				}
 				else {
-					Object bean = registry.getBean(name);
+					Object bean = registry.getBean(actualName);
 					// could be FunctionFactoryMetadata. . . TODO investigate and fix
 					if (bean instanceof FunctionFactoryMetadata) {
 						param = ((FunctionFactoryMetadata<?>) bean).getFactoryMethod().getGenericReturnType();
@@ -77,17 +88,6 @@ public abstract class FunctionContextUtils {
 		return param;
 	}
 
-//	private static Type findBeanType(AbstractBeanDefinition definition,
-//			MethodMetadataReadingVisitor visitor) {
-//		Class<?> factory = ClassUtils.resolveClassName(visitor.getDeclaringClassName(),
-//				null);
-//		Class<?>[] params = getParamTypes(factory, definition);
-//		Method method = ReflectionUtils.findMethod(factory, visitor.getMethodName(),
-//				params);
-//		Type type = method.getGenericReturnType();
-//		return type;
-//	}
-
 	private static Type findBeanType(AbstractBeanDefinition definition, String declaringClassName, String methodName) {
 		Class<?> factory = ClassUtils.resolveClassName(declaringClassName, null);
 		Class<?>[] params = getParamTypes(factory, definition);