GH-662 Fix support for reactive functions in AWS

This commit also includes other minor fixes around CustomRuntime which was getting in the way of this specific issue
 Added lookup for _HANDLER env variable
 Added few tests (will need more)
 Added support for Iterable for reactive functions

Resolves #662

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

@@ -32,6 +32,7 @@ import com.amazonaws.services.lambda.runtime.events.SNSEvent;
 import com.amazonaws.services.lambda.runtime.events.SQSEvent;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
 
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.context.annotation.Bean;
@@ -51,6 +52,8 @@ public class FunctionInvokerTests {
 
 	ObjectMapper mapper = new ObjectMapper();
 
+	String jsonCollection = "[\"Ricky\",\"Julien\",\"Bubbles\"]";
+
 	String sampleLBEvent = "{" +
 			"    \"requestContext\": {" +
 			"        \"elb\": {" +
@@ -360,6 +363,20 @@ public class FunctionInvokerTests {
 			"    \"isBase64Encoded\": false\n" +
 			"}";
 
+
+	@Test
+	public void testCollection() throws Exception {
+		System.setProperty("MAIN_CLASS", SampleConfiguration.class.getName());
+		System.setProperty("spring.cloud.function.definition", "echoStringReactive");
+		FunctionInvoker invoker = new FunctionInvoker();
+
+		InputStream targetStream = new ByteArrayInputStream(this.jsonCollection.getBytes());
+		ByteArrayOutputStream output = new ByteArrayOutputStream();
+		invoker.handleRequest(targetStream, output, null);
+		String result = new String(output.toByteArray(), StandardCharsets.UTF_8);
+		assertThat(result).isEqualTo(this.jsonCollection);
+	}
+
 	@Test
 	public void testKinesisStringEvent() throws Exception {
 		System.setProperty("MAIN_CLASS", KinesisConfiguration.class.getName());
@@ -689,6 +706,20 @@ public class FunctionInvokerTests {
 		assertThat(result.get("body")).isEqualTo("\"OK\"");
 	}
 
+	@EnableAutoConfiguration
+	@Configuration
+	public static class SampleConfiguration {
+		@Bean
+		public Function<String, String> echoString() {
+			return v -> v;
+		}
+
+		@Bean
+		public Function<Flux<String>, Flux<String>> echoStringReactive() {
+			return v -> v;
+		}
+	}
+
 	@EnableAutoConfiguration
 	@Configuration
 	public static class KinesisConfiguration {