Add support for different handler->function mappings in Azure

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

@@ -20,6 +20,7 @@ import java.lang.reflect.UndeclaredThrowableException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.function.Function;
 
 import com.microsoft.azure.functions.ExecutionContext;
 import com.microsoft.azure.functions.OutputBinding;
@@ -45,14 +46,15 @@ public class AzureSpringBootRequestHandler<I, O> extends AzureSpringFunctionInit
 		String name = null;
 		try {
 			if (context != null) {
-				context.getLogger().fine("Handler processed a request.");
 				name = context.getFunctionName();
+				context.getLogger().info("Handler processing a request for: " + name);
 			}
 			initialize(context);
 
-			Object convertedEvent = convertEvent(input);
-			Publisher<?> output = apply(name, extract(convertedEvent));
-			return result(convertedEvent, output);
+			Function<Publisher<?>, Publisher<?>> function = lookup(name);
+			Publisher<?> events = extract(function, convertEvent(input));
+			Publisher<?> output = function.apply(events);
+			return result(function, input, output);
 		}
 		catch (Throwable ex) {
 			if (context != null) {
@@ -68,7 +70,7 @@ public class AzureSpringBootRequestHandler<I, O> extends AzureSpringFunctionInit
 		}
 		finally {
 			if (context != null) {
-				context.getLogger().fine("Handler processed a request.");
+				context.getLogger().fine("Handler processed a request for: " + name);
 			}
 		}
 	}
@@ -83,19 +85,26 @@ public class AzureSpringBootRequestHandler<I, O> extends AzureSpringFunctionInit
 		return input;
 	}
 
-	private Flux<?> extract(Object input) {
-		if (input instanceof Collection) {
-			return Flux.fromIterable((Iterable<?>) input);
+	private Flux<?> extract(Function<?, ?> function, Object input) {
+		if (!isSingleInput(function, input)) {
+			if (input instanceof Collection) {
+				return Flux.fromIterable((Iterable<?>) input);
+			}
 		}
 		return Flux.just(input);
 	}
 
-	private O result(Object input, Publisher<?> output) {
+	private O result(Function<?, ?> function, Object input, Publisher<?> output) {
 		List<Object> result = new ArrayList<>();
 		for (Object value : Flux.from(output).toIterable()) {
 			result.add(convertOutput(value));
 		}
-		if (isSingleValue(input) && result.size() == 1) {
+		if (isSingleInput(function, input) && result.size() == 1) {
+			@SuppressWarnings("unchecked")
+			O value = (O) result.get(0);
+			return value;
+		}
+		if (isSingleOutput(function, input) && result.size() == 1) {
 			@SuppressWarnings("unchecked")
 			O value = (O) result.get(0);
 			return value;
@@ -105,10 +114,6 @@ public class AzureSpringBootRequestHandler<I, O> extends AzureSpringFunctionInit
 		return value;
 	}
 
-	private boolean isSingleValue(Object input) {
-		return !(input instanceof Collection);
-	}
-
 	@SuppressWarnings("unchecked")
 	protected O convertOutput(Object output) {
 		return (O) output;

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

@@ -20,6 +20,7 @@ import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URL;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
@@ -36,6 +37,7 @@ import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.function.context.FunctionCatalog;
+import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.util.ClassUtils;
 
@@ -53,6 +55,9 @@ public class AzureSpringFunctionInitializer implements Closeable {
 	@Autowired(required = false)
 	private FunctionCatalog catalog;
 
+	@Autowired(required = false)
+	private FunctionInspector inspector;
+
 	private volatile static ConfigurableApplicationContext context;
 
 	public AzureSpringFunctionInitializer(Class<?> configurationClass) {
@@ -68,6 +73,7 @@ public class AzureSpringFunctionInitializer implements Closeable {
 	public void close() throws IOException {
 		if (AzureSpringFunctionInitializer.context != null) {
 			AzureSpringFunctionInitializer.context.close();
+			AzureSpringFunctionInitializer.context = null;
 		}
 	}
 
@@ -80,7 +86,7 @@ public class AzureSpringFunctionInitializer implements Closeable {
 			return;
 		}
 		if (ctxt != null) {
-			ctxt.getLogger().info("Initializing function");
+			ctxt.getLogger().info("Initializing functions");
 		}
 
 		if (context == null) {
@@ -119,13 +125,37 @@ public class AzureSpringFunctionInitializer implements Closeable {
 		}
 	}
 
-	protected Publisher<?> apply(String name, Publisher<?> input) {
+	protected boolean isSingleInput(Function<?,?> function, Object input) {
+		if (!(input instanceof Collection)) {
+			return true;
+		}
+		if (this.inspector != null) {
+			return Collection.class.isAssignableFrom(inspector.getInputType(function));
+		}
+		return ((Collection<?>)input).size() <= 1;
+	}
+
+	protected boolean isSingleOutput(Function<?,?> function, Object output) {
+		if (!(output instanceof Collection)) {
+			return true;
+		}
+		if (this.inspector != null) {
+			return Collection.class.isAssignableFrom(inspector.getOutputType(function));
+		}
+		return ((Collection<?>)output).size() <= 1;
+	}
+
+	protected Function<Publisher<?>, Publisher<?>> lookup(String name) {
 		Function<Publisher<?>, Publisher<?>> function = this.function;
-		if (function == null && name != null) {
-			function = this.catalog.lookup(Function.class, name);
+		if (name != null && this.catalog != null) {
+			Function<Publisher<?>, Publisher<?>> preferred = this.catalog
+					.lookup(Function.class, name);
+			if (preferred != null) {
+				function = preferred;
+			}
 		}
 		if (function != null) {
-			return function.apply(input);
+			return function;
 		}
 		throw new IllegalStateException("No function defined");
 	}

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

@@ -0,0 +1,220 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.junit.After;
+import org.junit.Test;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class AzureSpringBootRequestHandlerTests {
+
+	private AzureSpringBootRequestHandler<?, ?> handler = null;
+
+	<I, O> AzureSpringBootRequestHandler<I, O> handler(Class<?> config) {
+		AzureSpringBootRequestHandler<I, O> handler = new AzureSpringBootRequestHandler<I, O>(
+				config);
+		this.handler = handler;
+		return handler;
+	}
+
+	@Test
+	public void bareConfig() {
+		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(BareConfig.class);
+		Bar bar = handler.handleRequest(new Foo("bar"),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar.getValue()).isEqualTo("BAR");
+	}
+
+	@Test
+	public void autoConfig() {
+		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(AutoConfig.class);
+		Bar bar = handler.handleRequest(new Foo("bar"),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar.getValue()).isEqualTo("BAR");
+	}
+
+	@Test
+	public void multiConfig() {
+		AzureSpringBootRequestHandler<Foo, Bar> handler = handler(MultiConfig.class);
+		Bar bar = handler.handleRequest(new Foo("bar"),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar.getValue()).isEqualTo("BAR");
+	}
+
+	@Test
+	public void implicitListConfig() {
+		AzureSpringBootRequestHandler<List<Foo>, List<Bar>> handler = handler(
+				AutoConfig.class);
+		List<Bar> bar = handler.handleRequest(Arrays.asList(new Foo("bar")),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar).hasSize(1);
+		assertThat(bar.get(0).getValue()).isEqualTo("BAR");
+	}
+
+	@Test
+	public void listToListConfig() {
+		AzureSpringBootRequestHandler<List<Foo>, List<Bar>> handler = handler(
+				ListConfig.class);
+		List<Bar> bar = handler.handleRequest(
+				Arrays.asList(new Foo("bar"), new Foo("baz")),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar).hasSize(2);
+		assertThat(bar.get(0).getValue()).isEqualTo("BAR");
+	}
+
+	@Test
+	public void listToListSingleConfig() {
+		AzureSpringBootRequestHandler<List<Foo>, List<Bar>> handler = handler(
+				ListConfig.class);
+		List<Bar> bar = handler.handleRequest(Arrays.asList(new Foo("bar")),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar).hasSize(1);
+		assertThat(bar.get(0).getValue()).isEqualTo("BAR");
+	}
+
+	@Test
+	public void collectConfig() {
+		AzureSpringBootRequestHandler<List<Foo>, Bar> handler = handler(CollectConfig.class);
+		Bar bar = handler.handleRequest(Arrays.asList(new Foo("bar")),
+				new TestExecutionContext("uppercase"));
+		assertThat(bar.getValue()).isEqualTo("BAR");
+	}
+
+	@After
+	public void close() throws IOException {
+		if (handler != null)
+			handler.close();
+	}
+
+	@Configuration
+	protected static class BareConfig {
+		@Bean
+		public Function<Flux<Foo>, Flux<Bar>> function() {
+			return foos -> foos.map(foo -> new Bar(foo.getValue().toUpperCase()));
+		}
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	protected static class AutoConfig {
+		@Bean
+		public Function<Foo, Bar> uppercase() {
+			return foo -> new Bar(foo.getValue().toUpperCase());
+		}
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	protected static class ListConfig {
+		@Bean
+		public Function<List<Foo>, List<Bar>> uppercase() {
+			return foos -> foos.stream().map(foo -> new Bar(foo.getValue().toUpperCase()))
+					.collect(Collectors.toList());
+		}
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	protected static class CollectConfig {
+		@Bean
+		public Function<List<Foo>, Bar> uppercase() {
+			return foos -> new Bar(foos.stream().map(foo -> foo.getValue().toUpperCase())
+					.collect(Collectors.joining(",")));
+		}
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	protected static class MultiConfig {
+		@Bean
+		public Function<Foo, Bar> uppercase() {
+			return foo -> new Bar(foo.getValue().toUpperCase());
+		}
+
+		@Bean
+		public Function<Bar, Foo> lowercase() {
+			return bar -> new Foo(bar.getValue().toLowerCase());
+		}
+	}
+
+}
+
+class Foo {
+
+	private String value;
+
+	Foo() {
+	}
+
+	public String lowercase() {
+		return value.toLowerCase();
+	}
+
+	public Foo(String value) {
+		this.value = value;
+	}
+
+	public String uppercase() {
+		return value.toUpperCase();
+	}
+
+	public String getValue() {
+		return value;
+	}
+
+	public void setValue(String value) {
+		this.value = value;
+	}
+}
+
+class Bar {
+
+	private String value;
+
+	Bar() {
+	}
+
+	public Bar(String value) {
+		this.value = value;
+	}
+
+	public String getValue() {
+		return value;
+	}
+
+	public void setValue(String value) {
+		this.value = value;
+	}
+
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.util.UUID;
+import java.util.logging.Logger;
+
+import com.microsoft.azure.functions.ExecutionContext;
+
+public class TestExecutionContext implements ExecutionContext {
+	private String name;
+
+	public TestExecutionContext(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public Logger getLogger() {
+		return Logger.getAnonymousLogger();
+	}
+
+	@Override
+	public String getInvocationId() {
+		return UUID.randomUUID().toString();
+	}
+
+	@Override
+	public String getFunctionName() {
+		return name;
+	}
+}