GH-1801 Fix NPE when factory method is null

Fixed NPE for cases when factory method is null by attemptimng to discover factory method via source.MethodMetadata
Added tests

Resolves #1801

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -41,6 +41,7 @@ import org.springframework.cloud.function.context.PollableSupplier;
 import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
+import org.springframework.cloud.function.context.config.FunctionContextUtils;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.BindingCreatedEvent;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
@@ -54,6 +55,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.type.MethodMetadata;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
@@ -66,8 +68,10 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Oleg Zhurakousky
@@ -95,7 +99,7 @@ public class FunctionConfiguration {
 		IntegrationFlow integrationFlow = null;
 		if (functionCatalog != null && ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class))) {
 			FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionProperties.getDefinition());
-			if (functionWrapper != null) {
+			if (functionWrapper != null /*&& functionWrapper.getTarget() instanceof Supplier*/) {
 				AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
 				Publisher<Object> beginPublishingTrigger = Mono.create(emmiter -> {
 					triggerRef.set(emmiter);
@@ -108,9 +112,18 @@ public class FunctionConfiguration {
 					}
 				});
 
-
 				RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]);
 				Method factoryMethod = bd.getResolvedFactoryMethod();
+				if (factoryMethod == null) {
+					Object source = bd.getSource();
+					if (source instanceof MethodMetadata) {
+						Class<?> factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null);
+						Class<?>[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, bd);
+						factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params);
+					}
+				}
+				Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '"
+								+ functionProperties.getDefinition() + "'");
 				PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class)
 						? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class)
 								: null;
@@ -158,7 +171,6 @@ public class FunctionConfiguration {
 		return value instanceof Message ? (Message<T>) value : MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build();
 	}
 
-
 	/**
 	 *
 	 * @author Oleg Zhurakousky

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/edgecases/GH1801Test.java

@@ -0,0 +1,59 @@
+/*
+ * 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.stream.function.edgecases;
+
+import java.util.function.Consumer;
+
+import org.junit.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+
+/**
+ * This test validates that the issue https://github.com/spring-cloud/spring-cloud-stream/issues/1801
+ * is addressed.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class GH1801Test {
+
+	@Test
+	public void test() {
+		SampleBootApplication.main("--spring.cloud.stream.defaultBinder=integration");
+	}
+
+	@SpringBootApplication
+	public static class SampleBootApplication {
+		public static void main(String... args) {
+			new SpringApplicationBuilder(SampleBootApplication.class).web(WebApplicationType.NONE).run(args);
+		}
+	}
+}
+
+@Configuration
+class StreamConfiguration {
+
+	@Bean
+	public Consumer<Message<?>> consumer() {
+		return System.out::println;
+	}
+}