From 71f3af825d75d6a288a520bb48078dece78bf99b Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Thu, 18 Nov 2021 13:29:07 +0100
Subject: [PATCH] GH-727 Fix Azure's Supplier<Publisher> logic to avoid NPE

Resolves #727
---
 .../adapter/azure/FunctionInvoker.java        | 22 ++++++++++------
 .../adapter/azure/FunctionInvokerTests.java   | 26 +++++++++++++++++++
 2 files changed, 40 insertions(+), 8 deletions(-)

diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/FunctionInvoker.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/FunctionInvoker.java
index 01c9f5911..ee1f1dd07 100644
--- a/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/FunctionInvoker.java
+++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/FunctionInvoker.java
@@ -110,7 +110,11 @@ public class FunctionInvoker<I, O> {
 	public O handleRequest(I input, ExecutionContext executionContext) {
 		String functionDefinition = executionContext.getFunctionName();
 		FunctionInvocationWrapper function = FUNCTION_CATALOG.lookup(functionDefinition);
-		if (function == null && StringUtils.hasText(functionDefinition) && APPLICATION_CONTEXT.containsBean(functionDefinition)) {
+		if (function != null && StringUtils.hasText(functionDefinition) && !function.getFunctionDefinition().equals(functionDefinition)) {
+			this.registerFunction(functionDefinition);
+			function = FUNCTION_CATALOG.lookup(functionDefinition);
+		}
+		else if (function == null && StringUtils.hasText(functionDefinition) && APPLICATION_CONTEXT.containsBean(functionDefinition)) {
 			this.registerFunction(functionDefinition);
 			function = FUNCTION_CATALOG.lookup(functionDefinition);
 		}
@@ -128,7 +132,7 @@ public class FunctionInvoker<I, O> {
 						resultList.addAll((Collection) resultItem);
 					}
 					else {
-						if (Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getInputType()))
+						if (!function.isSupplier() && Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getInputType()))
 								&& !Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getOutputType()))) {
 							return (O) this.convertOutputIfNecessary(input, resultItem);
 						}
@@ -145,15 +149,17 @@ public class FunctionInvoker<I, O> {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	private void registerFunction(String functionDefinition) {
-		FunctionRegistration functionRegistration =
-				new FunctionRegistration(APPLICATION_CONTEXT.getBean(functionDefinition), functionDefinition);
+		if (APPLICATION_CONTEXT.containsBean(functionDefinition)) {
+			FunctionRegistration functionRegistration =
+					new FunctionRegistration(APPLICATION_CONTEXT.getBean(functionDefinition), functionDefinition);
 
-		Type type = FunctionContextUtils.
-				findType(functionDefinition, APPLICATION_CONTEXT.getBeanFactory());
+			Type type = FunctionContextUtils.
+					findType(functionDefinition, APPLICATION_CONTEXT.getBeanFactory());
 
 		functionRegistration = functionRegistration.type(type);
 
-		((FunctionRegistry) FUNCTION_CATALOG).register(functionRegistration);
+			((FunctionRegistry) FUNCTION_CATALOG).register(functionRegistration);
+		}
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -186,7 +192,7 @@ public class FunctionInvoker<I, O> {
 
 	@SuppressWarnings("unchecked")
 	private Object convertOutputIfNecessary(Object input, Object output) {
-		if (input != null && input instanceof HttpRequestMessage) {
+		if (input instanceof HttpRequestMessage) {
 			HttpRequestMessage<I> requestMessage = (HttpRequestMessage<I>) input;
 			Map<String, Object> headers = null;
 			if (output instanceof Message) {
diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/FunctionInvokerTests.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/FunctionInvokerTests.java
index 1f31d49ce..cc024d7ea 100644
--- a/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/FunctionInvokerTests.java
+++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/FunctionInvokerTests.java
@@ -134,6 +134,16 @@ public class FunctionInvokerTests {
 		assertThat(result.toString()).isEqualTo("[foo1, foo2]");
 	}
 
+	@Test
+	public void supplierPublisherBean() {
+		FunctionInvoker<Void, ?> handler = handler(ReactiveSupplierConfig.class);
+		Foo resultSingle = (Foo) handler.handleRequest(new TestExecutionContext("suppliermono"));
+		assertThat(resultSingle.getValue()).isEqualTo("hello");
+
+		List<Foo> resultList = (List<Foo>) handler.handleRequest(new TestExecutionContext("supplierflux"));
+		assertThat(resultList.size()).isEqualTo(2);
+	}
+
 	private static String consumerResult;
 
 	@Test
@@ -209,6 +219,22 @@ public class FunctionInvokerTests {
 
 	}
 
+	@Configuration
+//	@EnableAutoConfiguration
+	protected static class ReactiveSupplierConfig {
+
+		@Bean
+		public Supplier<Mono<Foo>> suppliermono() {
+			return () -> Mono.just(new Foo("hello"));
+		}
+
+		@Bean
+		public Supplier<Flux<Foo>> supplierflux() {
+			return () -> Flux.just(new Foo("hello"), new Foo("bye"));
+		}
+
+	}
+
 	@Configuration
 	protected static class BareConfig {