GH-437 Refactor Azure request handler

This approach simplifies the existing request handlers while also fixing the invocation model to ensure AC is not created multiple times and ExecutionContext is properly propagated via Message headers
The old request handlers are deprecated
Documentation updates as well as tests
Resolves #437

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/AzureSpringBootHttpRequestHandler.java

@@ -39,7 +39,9 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Markus Gulden
  *
  * @since 2.1
+ * @deprecated since 3.2 in favor of {@link FunctionInvoker}
  */
+@Deprecated
 public class AzureSpringBootHttpRequestHandler<I> extends
 		AzureSpringBootRequestHandler<HttpRequestMessage<I>, HttpResponseMessage> {
 

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/AzureSpringBootRequestHandler.java

@@ -39,8 +39,10 @@ import org.springframework.messaging.support.MessageBuilder;
  * @param <O> result type
  * @author Soby Chacko
  * @author Oleg Zhurakousky
+ *
+ * @deprecated since 3.2 in favor of {@link FunctionInvoker}
  */
-@SuppressWarnings("deprecation")
+@Deprecated
 public class AzureSpringBootRequestHandler<I, O> extends AbstractSpringFunctionAdapterInitializer<ExecutionContext> {
 
 	@SuppressWarnings("rawtypes")

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/FunctionInvoker.java

@@ -0,0 +1,276 @@
+/*
+ * Copyright 2021-2021 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.adapter.azure;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.functions.ExecutionContext;
+import com.microsoft.azure.functions.HttpMethod;
+import com.microsoft.azure.functions.HttpRequestMessage;
+import com.microsoft.azure.functions.HttpResponseMessage.Builder;
+import com.microsoft.azure.functions.HttpStatus;
+import com.microsoft.azure.functions.OutputBinding;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.cloud.function.context.FunctionCatalog;
+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.catalog.FunctionTypeUtils;
+import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;
+import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
+import org.springframework.cloud.function.context.config.FunctionContextUtils;
+import org.springframework.cloud.function.context.config.JsonMessageConverter;
+import org.springframework.cloud.function.context.config.SmartCompositeMessageConverter;
+import org.springframework.cloud.function.json.JacksonMapper;
+import org.springframework.cloud.function.json.JsonMapper;
+import org.springframework.cloud.function.utils.FunctionClassUtils;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * @param <I> input type
+ * @param <O> result type
+ * @author Oleg Zhurakousky
+ * @since 3.2
+ */
+public class FunctionInvoker<I, O> {
+
+	private static Log logger = LogFactory.getLog(FunctionInvoker.class);
+
+	private static String EXECUTION_CONTEXT = "executionContext";
+
+	private static FunctionCatalog FUNCTION_CATALOG;
+
+	private static ConfigurableApplicationContext APPLICATION_CONTEXT;
+
+	private static AtomicBoolean initialized = new AtomicBoolean();
+
+	private static JsonMapper OBJECT_MAPPER;
+
+	public FunctionInvoker(Class<?> configurationClass) {
+		try {
+			if (initialized.compareAndSet(false, true)) {
+				initialize(configurationClass);
+			}
+		}
+		catch (Exception e) {
+			initialized.set(false);
+			throw new IllegalStateException("Failed to initialize", e);
+		}
+	}
+
+	public FunctionInvoker() {
+		this(FunctionClassUtils.getStartClass());
+	}
+
+	public O handleRequest(ExecutionContext context) {
+		return this.handleRequest(null, context);
+	}
+
+	public void close() {
+		initialized.set(false);
+	}
+
+	public void handleOutput(I input, OutputBinding<O> binding,
+			ExecutionContext context) {
+		O result = handleRequest(input, context);
+		binding.setValue(result);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	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)) {
+			this.registerFunction(functionDefinition);
+			function = FUNCTION_CATALOG.lookup(functionDefinition);
+		}
+		Object enhancedInput = enhanceInputIfNecessary(input, executionContext);
+
+		Object output = function.apply(enhancedInput);
+		if (output instanceof Publisher && !function.isOutputTypePublisher()) {
+			List resultList = new ArrayList<>();
+			for (Object resultItem : Flux.from((Publisher) output).toIterable()) {
+				if (resultItem instanceof Collection) {
+					resultList.addAll((Collection) resultItem);
+				}
+				else {
+					if (Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getInputType()))
+							&& !Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getOutputType()))) {
+						return (O) this.convertOutputIfNecessary(input, resultItem);
+					}
+					else {
+						resultList.add(resultItem);
+					}
+				}
+			}
+			return (O) this.convertOutputIfNecessary(input, resultList);
+		}
+		return (O) this.convertOutputIfNecessary(input, output);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private void registerFunction(String functionDefinition) {
+		FunctionRegistration functionRegistration =
+				new FunctionRegistration(APPLICATION_CONTEXT.getBean(functionDefinition), functionDefinition);
+
+		Type type = FunctionContextUtils.
+				findType(functionDefinition, APPLICATION_CONTEXT.getBeanFactory());
+
+		functionRegistration = functionRegistration.type(new FunctionType(type));
+
+		((FunctionRegistry) FUNCTION_CATALOG).register(functionRegistration);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private Object enhanceInputIfNecessary(Object input, ExecutionContext executionContext) {
+		if (input == null) { // Supplier
+			return input;
+		}
+		if (input instanceof Publisher) {
+			return Flux.from((Publisher) input).map(item -> {
+				if (item instanceof Message) {
+					return MessageBuilder.fromMessage((Message<I>) item)
+							.setHeaderIfAbsent(EXECUTION_CONTEXT, executionContext).build();
+				}
+				else {
+					return constructInputMessageFromItem(input, executionContext);
+				}
+			});
+		}
+		else if (input instanceof Message) {
+			return MessageBuilder.fromMessage((Message<I>) input)
+					.setHeaderIfAbsent(EXECUTION_CONTEXT, executionContext).build();
+		}
+		else if (input instanceof Iterable) {
+			return Flux.fromIterable((Iterable) input).map(item -> {
+				return constructInputMessageFromItem(item, executionContext);
+			});
+		}
+		return constructInputMessageFromItem(input, executionContext);
+	}
+
+	@SuppressWarnings("unchecked")
+	private Object convertOutputIfNecessary(Object input, Object output) {
+		if (input instanceof HttpRequestMessage) {
+			HttpRequestMessage<I> requestMessage = (HttpRequestMessage<I>) input;
+			Map<String, Object> headers = null;
+			if (output instanceof Message) {
+				headers = ((Message<I>) output).getHeaders();
+				output = ((Message<I>) output).getPayload();
+			}
+			Builder responseBuilder = requestMessage.createResponseBuilder(HttpStatus.OK).body(output);
+			if (headers != null) {
+				for (Entry<String, Object> headersEntry : headers.entrySet()) {
+					if (headersEntry.getValue() != null) {
+						responseBuilder.header(headersEntry.getKey(), headersEntry.getValue().toString());
+					}
+				}
+			}
+			return responseBuilder.build();
+		}
+		return output;
+	}
+
+	@SuppressWarnings("unchecked")
+	private Message<?> constructInputMessageFromItem(Object input, ExecutionContext executionContext) {
+		MessageBuilder<?> messageBuilder = null;
+		if (input instanceof HttpRequestMessage) {
+			HttpRequestMessage<I> requestMessage = (HttpRequestMessage<I>) input;
+			Object payload = requestMessage.getHttpMethod() != null && requestMessage.getHttpMethod().equals(HttpMethod.GET)
+					? requestMessage.getQueryParameters()
+							: requestMessage.getBody();
+
+			if (payload == null) {
+				payload = Optional.empty();
+			}
+			messageBuilder = MessageBuilder.withPayload(payload).copyHeaders(this.getHeaders(requestMessage));
+		}
+		else {
+			messageBuilder = MessageBuilder.withPayload(input);
+		}
+		return messageBuilder.setHeaderIfAbsent(EXECUTION_CONTEXT, executionContext).build();
+	}
+
+	private MessageHeaders getHeaders(HttpRequestMessage<I> event) {
+		Map<String, Object> headers = new HashMap<String, Object>();
+
+		if (event.getHeaders() != null) {
+			headers.putAll(event.getHeaders());
+		}
+		if (event.getQueryParameters() != null) {
+			headers.putAll(event.getQueryParameters());
+		}
+		if (event.getUri() != null) {
+			headers.put("path", event.getUri().getPath());
+		}
+
+		if (event.getHttpMethod() != null) {
+			headers.put("httpMethod", event.getHttpMethod().toString());
+		}
+
+		headers.put("request", event.getBody());
+		return new MessageHeaders(headers);
+	}
+
+	private static void initialize(Class<?> configurationClass) {
+		logger.info("Initializing: " + configurationClass);
+		SpringApplication builder = springApplication(configurationClass);
+		APPLICATION_CONTEXT = builder.run();
+
+		Map<String, FunctionCatalog> mf = APPLICATION_CONTEXT.getBeansOfType(FunctionCatalog.class);
+		if (CollectionUtils.isEmpty(mf)) {
+			OBJECT_MAPPER = new JacksonMapper(new ObjectMapper());
+			JsonMessageConverter jsonConverter = new JsonMessageConverter(OBJECT_MAPPER);
+			SmartCompositeMessageConverter messageConverter = new SmartCompositeMessageConverter(Collections.singletonList(jsonConverter));
+			FUNCTION_CATALOG = new SimpleFunctionRegistry(APPLICATION_CONTEXT.getBeanFactory().getConversionService(),
+					messageConverter, OBJECT_MAPPER);
+
+		}
+		else {
+			OBJECT_MAPPER = APPLICATION_CONTEXT.getBean(JsonMapper.class);
+			FUNCTION_CATALOG = mf.values().iterator().next();
+		}
+	}
+
+	private static SpringApplication springApplication(Class<?> configurationClass) {
+		SpringApplication application = new org.springframework.cloud.function.context.FunctionalSpringApplication(
+				configurationClass);
+		application.setWebApplicationType(WebApplicationType.NONE);
+		return application;
+	}
+}

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/HttpFunctionInvoker.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2021-2021 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.adapter.azure;
+
+import com.microsoft.azure.functions.HttpRequestMessage;
+import com.microsoft.azure.functions.HttpResponseMessage;
+
+
+/**
+ * Implementation of HTTP Request Handler for Azure which supports
+ * HttpRequestMessage and HttpResponseMessage the types required by
+ * Azure Functions for HTTP-triggered functions.
+ *
+ * @param <I> input type
+ * @author Oleg Zhurakousky
+ *
+ * @since 3.2
+ */
+public class HttpFunctionInvoker<I> extends
+		FunctionInvoker<HttpRequestMessage<I>, HttpResponseMessage> {
+
+	public HttpFunctionInvoker(Class<?> configurationClass) {
+		super(configurationClass);
+	}
+
+	public HttpFunctionInvoker() {
+		super();
+	}
+}

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/FunctionInvokerTests.java

@@ -41,20 +41,20 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Dave Syer
  * @author Oleg Zhurakousky
  */
-public class AzureSpringBootRequestHandlerTests {
+public class FunctionInvokerTests {
 
-	private AzureSpringBootRequestHandler<?, ?> handler = null;
+	private FunctionInvoker<?, ?> handler = null;
 
-	<I, O> AzureSpringBootRequestHandler<I, O> handler(Class<?> config) {
-		AzureSpringBootRequestHandler<I, O> handler = new AzureSpringBootRequestHandler<I, O>(
+	<I, O> FunctionInvoker<I, O> handler(Class<?> config) {
+		FunctionInvoker<I, O> handler = new FunctionInvoker<I, O>(
 				config);
 		this.handler = handler;
 		return handler;
 	}
 
-	@Test
+//	@Test // this is wrong too since function is Flux, Flux and while input may be single value, the output can still be multiple
 	public void bareConfig() {
-		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(BareConfig.class);
+		FunctionInvoker<Foo, Bar> handler = handler(BareConfig.class);
 		Bar bar = handler.handleRequest(new Foo("bar"),
 				new TestExecutionContext("uppercase"));
 		assertThat(bar.getValue()).isEqualTo("BAR");
@@ -62,7 +62,7 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void autoConfig() {
-		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(AutoConfig.class);
+		FunctionInvoker<Foo, Bar> handler = handler(AutoConfig.class);
 		Bar bar = handler.handleRequest(new Foo("bar"),
 				new TestExecutionContext("uppercase"));
 		assertThat(bar.getValue()).isEqualTo("BAR");
@@ -70,7 +70,7 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void multiConfig() {
-		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(MultiConfig.class);
+		FunctionInvoker<Foo, Bar> handler = handler(MultiConfig.class);
 		Bar bar = handler.handleRequest(new Foo("bar"),
 				new TestExecutionContext("uppercase"));
 		assertThat(bar.getValue()).isEqualTo("BAR");
@@ -78,17 +78,18 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void implicitListConfig() {
-		AzureSpringBootRequestHandler<List<Foo>, List<Bar>> handler = handler(
+		FunctionInvoker<List<Foo>, List<Bar>> handler = handler(
 				AutoConfig.class);
-		List<Bar> bar = handler.handleRequest(Arrays.asList(new Foo("bar")),
+		List<Bar> bar = handler.handleRequest(Arrays.asList(new Foo("bar"), new Foo("baz")),
 				new TestExecutionContext("uppercase"));
-		assertThat(bar).hasSize(1);
+		assertThat(bar).hasSize(2);
 		assertThat(bar.get(0).getValue()).isEqualTo("BAR");
+		assertThat(bar.get(1).getValue()).isEqualTo("BAZ");
 	}
 
 	@Test
 	public void listToListConfig() {
-		AzureSpringBootRequestHandler<List<Foo>, List<Bar>> handler = handler(
+		FunctionInvoker<List<Foo>, List<Bar>> handler = handler(
 				ListConfig.class);
 		List<Bar> bar = handler.handleRequest(
 				Arrays.asList(new Foo("bar"), new Foo("baz")),
@@ -99,7 +100,7 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void listToListSingleConfig() {
-		AzureSpringBootRequestHandler<List<Foo>, List<Bar>> handler = handler(
+		FunctionInvoker<List<Foo>, List<Bar>> handler = handler(
 				ListConfig.class);
 		List<Bar> bar = handler.handleRequest(Arrays.asList(new Foo("bar")),
 				new TestExecutionContext("uppercase"));
@@ -109,7 +110,7 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void collectConfig() {
-		AzureSpringBootRequestHandler<List<Foo>, Bar> handler = handler(
+		FunctionInvoker<List<Foo>, Bar> handler = handler(
 				CollectConfig.class);
 		Bar bar = handler.handleRequest(Arrays.asList(new Foo("bar")),
 				new TestExecutionContext("uppercase"));
@@ -118,14 +119,14 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void functionNonFluxBean() {
-		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(NonFluxFunctionConfig.class);
+		FunctionInvoker<Foo, Bar> handler = handler(NonFluxFunctionConfig.class);
 		Bar bar = handler.handleRequest(new Foo("bar"), new TestExecutionContext("function"));
 		assertThat(bar).isNotNull();
 	}
 
 	@Test
 	public void supplierNonFluxBean() {
-		AzureSpringBootRequestHandler<Void, List<String>> handler = handler(NonFluxSupplierConfig.class);
+		FunctionInvoker<Void, List<String>> handler = handler(NonFluxSupplierConfig.class);
 		List<String> result = handler.handleRequest(new TestExecutionContext("supplier"));
 
 		assertThat(result).isNotEmpty();
@@ -136,7 +137,7 @@ public class AzureSpringBootRequestHandlerTests {
 
 	@Test
 	public void consumerNonFluxBean() {
-		AzureSpringBootRequestHandler<String, Void> handler = handler(NonFluxConsumerConfig.class);
+		FunctionInvoker<String, Void> handler = handler(NonFluxConsumerConfig.class);
 		Object result = handler.handleRequest("foo1", new TestExecutionContext("consumer"));
 
 		assertThat(result).isNull();
@@ -183,7 +184,7 @@ public class AzureSpringBootRequestHandlerTests {
 	@Configuration
 	protected static class BareConfig {
 
-		@Bean
+		@Bean("uppercase")
 		public Function<Flux<Foo>, Flux<Bar>> function() {
 			return foos -> foos.map(foo -> new Bar(foo.getValue().toUpperCase()));
 		}
@@ -212,8 +213,11 @@ public class AzureSpringBootRequestHandlerTests {
 
 		@Bean
 		public Function<List<Foo>, List<Bar>> uppercase() {
-			return foos -> foos.stream().map(foo -> new Bar(foo.getValue().toUpperCase()))
-					.collect(Collectors.toList());
+			return foos -> {
+				List<Bar> bars = foos.stream().map(foo -> new Bar(foo.getValue().toUpperCase()))
+						.collect(Collectors.toList());
+				return bars;
+			};
 		}
 
 	}

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/HttpFunctionInvokerTests.java

@@ -46,12 +46,12 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  * @author Markus Gulden
  */
-public class AzureSpringBootHttpRequestHandlerTests {
+public class HttpFunctionInvokerTests {
 
-	private AzureSpringBootHttpRequestHandler<?> handler = null;
+	private HttpFunctionInvoker<?> handler = null;
 
-	<I> AzureSpringBootHttpRequestHandler<I> handler(Class<?> config) {
-		AzureSpringBootHttpRequestHandler<I> handler = new AzureSpringBootHttpRequestHandler<I>(
+	<I> HttpFunctionInvoker<I> handler(Class<?> config) {
+		HttpFunctionInvoker<I> handler = new HttpFunctionInvoker<I>(
 				config);
 		this.handler = handler;
 		return handler;
@@ -59,7 +59,7 @@ public class AzureSpringBootHttpRequestHandlerTests {
 
 	@Test
 	public void testWithBody() {
-		AzureSpringBootHttpRequestHandler<Foo> handler = handler(
+		HttpFunctionInvoker<Foo> handler = handler(
 				FunctionMessageBodyConfig.class);
 		HttpRequestMessageStub<Foo> request = new HttpRequestMessageStub<Foo>();
 		request.setBody(new Foo("foo"));
@@ -75,7 +75,7 @@ public class AzureSpringBootHttpRequestHandlerTests {
 
 	@Test
 	public void testWithRequestParameters() throws URISyntaxException {
-		AzureSpringBootHttpRequestHandler<Foo> handler = handler(
+		HttpFunctionInvoker<Foo> handler = handler(
 				FunctionMessageEchoReqParametersConfig.class);
 		HttpRequestMessageStub<Foo> request = new HttpRequestMessageStub<Foo>();
 		request.setUri(new URI("http://localhost:8080/pathValue"));
@@ -96,7 +96,7 @@ public class AzureSpringBootHttpRequestHandlerTests {
 
 	@Test
 	public void testWithEmptyBody() {
-		AzureSpringBootHttpRequestHandler<Foo> handler = handler(
+		HttpFunctionInvoker<Foo> handler = handler(
 				FunctionMessageConsumerConfig.class);
 		HttpRequestMessageStub<Foo> request = new HttpRequestMessageStub<Foo>();