add adapters as child module

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/FunctionInvokingKinesisEventHandler.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2017 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.aws;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Date;
+
+import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
+import com.amazonaws.services.lambda.runtime.events.KinesisEvent.KinesisEventRecord;
+
+/**
+ * @author Mark Fisher
+ */
+public class FunctionInvokingKinesisEventHandler extends SpringBootRequestHandler<KinesisEvent, String> {
+
+	public FunctionInvokingKinesisEventHandler(Class<?> configurationClass) {
+		super(configurationClass);
+	}
+
+	@Override
+	protected String convertEvent(KinesisEvent event) {
+		StringBuilder result = new StringBuilder();
+		for (KinesisEventRecord record : event.getRecords()) {
+			String id = record.getEventID();
+			String name = record.getEventName();
+			String source = record.getEventSource();
+			Date timestamp = record.getKinesis().getApproximateArrivalTimestamp();
+			String partitionKey = record.getKinesis().getPartitionKey();
+			String sequenceNumber = record.getKinesis().getSequenceNumber();
+			ByteBuffer data = record.getKinesis().getData();
+			String dataString = new String(data.array(), Charset.forName("UTF-8"));
+			result.append(String.format("id=%s,name=%s,source=%s,timestamp=%s,partitionKey=%s,sequenceNumber=%s,data=%s",
+					id, name, source, timestamp, partitionKey, sequenceNumber, dataString));
+		}
+		return result.toString();
+	}
+}

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/FunctionInvokingS3EventHandler.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017 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.aws;
+
+import com.amazonaws.services.lambda.runtime.events.S3Event;
+
+/**
+ * @author Mark Fisher
+ */
+public class FunctionInvokingS3EventHandler extends SpringBootRequestHandler<S3Event, String> {
+
+	public FunctionInvokingS3EventHandler(Class<?> configurationClass) {
+		super(configurationClass);
+	}
+
+	@Override
+	protected String convertEvent(S3Event event) {
+		return event.toJson();
+	}
+}

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringBootRequestHandler.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2017 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.aws;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Mark Fisher
+ */
+public class SpringBootRequestHandler<E, O> extends SpringFunctionInitializer implements RequestHandler<E, List<O>> {
+
+	public SpringBootRequestHandler(Class<?> configurationClass) {
+		super(configurationClass);
+	}
+
+	public SpringBootRequestHandler() {
+		super();
+	}
+
+	@Override
+	public List<O> handleRequest(E event, Context context) {
+		initialize();
+		Object input = convertEvent(event);
+		Flux<?> output = apply(extract(input));
+		return result(output);
+	}
+
+	private List<O> result(Flux<?> output) {
+		List<Object> result = new ArrayList<>();
+		for (Object value : output.toIterable()) {
+			result.add(value);
+		}
+		return convertResult(result);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected List<O> convertResult(List<Object> value) {
+		return (List<O>) value;
+	}
+
+	private Flux<?> extract(Object input) {
+		if (input instanceof Collection) {
+			return Flux.fromIterable((Iterable<?>) input);
+		}
+		return Flux.just(input);
+	}
+
+	protected Object convertEvent(E event) {
+		return event;
+	}
+
+}

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringBootStreamHandler.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2017 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.aws;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.springframework.beans.factory.annotation.Autowired;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ */
+public class SpringBootStreamHandler extends SpringFunctionInitializer
+		implements RequestStreamHandler {
+
+	@Autowired
+	private ObjectMapper mapper;
+
+	public SpringBootStreamHandler() {
+		super();
+	}
+
+	public SpringBootStreamHandler(Class<?> configurationClass) {
+		super(configurationClass);
+	}
+
+	@Override
+	public void handleRequest(InputStream input, OutputStream output, Context context)
+			throws IOException {
+		initialize();
+		Object value = convertStream(input);
+		Flux<?> flux = apply(extract(value));
+		mapper.writeValue(output, result(flux));
+	}
+
+	private Object result(Flux<?> flux) {
+		List<Object> result = new ArrayList<>();
+		for (Object value : flux.toIterable()) {
+			result.add(value);
+		}
+		return result;
+	}
+
+	private Flux<?> extract(Object input) {
+		if (input instanceof Collection) {
+			return Flux.fromIterable((Iterable<?>) input);
+		}
+		return Flux.just(input);
+	}
+
+	private Object convertStream(InputStream input) {
+		try {
+			return mapper.readValue(input, getInputType());
+		}
+		catch (Exception e) {
+			throw new IllegalStateException("Cannot convert event", e);
+		}
+	}
+}

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringFunctionInitializer.java

@@ -0,0 +1,191 @@
+/*
+ * Copyright 2017 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.aws;
+
+import java.io.Closeable;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.jar.Manifest;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.function.context.FunctionInspector;
+import org.springframework.cloud.function.registry.FunctionCatalog;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.util.ClassUtils;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ */
+public class SpringFunctionInitializer implements Closeable {
+
+	private static Log logger = LogFactory.getLog(SpringFunctionInitializer.class);
+
+	private final Class<?> configurationClass;
+
+	private Function<Flux<?>, Flux<?>> function;
+
+	private Consumer<Flux<?>> consumer;
+
+	private Supplier<Flux<?>> supplier;
+
+	private AtomicBoolean initialized = new AtomicBoolean();
+
+	@Autowired(required = false)
+	private FunctionInspector inspector;
+
+	@Autowired(required = false)
+	private FunctionCatalog catalog;
+
+	private String name;
+
+	private ConfigurableApplicationContext context;
+
+	public SpringFunctionInitializer(Class<?> configurationClass) {
+		this.configurationClass = configurationClass;
+	}
+
+	public SpringFunctionInitializer() {
+		this(getStartClass());
+	}
+
+	@Override
+	public void close() {
+		if (this.context != null) {
+			this.context.close();
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	protected void initialize() {
+		if (!this.initialized.compareAndSet(false, true)) {
+			return;
+		}
+		logger.info("Initializing: " + configurationClass);
+		SpringApplicationBuilder builder = new SpringApplicationBuilder(
+				configurationClass);
+		ConfigurableApplicationContext context = builder.web(false).run();
+		context.getAutowireCapableBeanFactory().autowireBean(this);
+		String name = context.getEnvironment().getProperty("function.name");
+		boolean defaultName = false;
+		if (name == null) {
+			name = "function";
+			defaultName = true;
+		}
+		if (this.catalog == null) {
+			this.function = context.getBean(name, Function.class);
+		}
+		else {
+			this.function = this.catalog.lookupFunction(name);
+			this.name = name;
+			if (this.function == null) {
+				if (defaultName) {
+					name = "consumer";
+				}
+				this.consumer = this.catalog.lookupConsumer(name);
+				this.name = name;
+				if (this.consumer == null) {
+					if (defaultName) {
+						name = "supplier";
+					}
+					this.supplier = this.catalog.lookupSupplier(name);
+					this.name = name;
+				}
+			}
+		}
+		this.context = context;
+	}
+
+	protected Class<?> getInputType() {
+		if (inspector != null) {
+			return inspector.getInputType(this.name);
+		}
+		return Object.class;
+	}
+
+	protected Flux<?> apply(Flux<?> input) {
+		if (this.function != null) {
+			return function.apply(input);
+		}
+		if (this.consumer != null) {
+			this.consumer.accept(input);
+			return Flux.empty();
+		}
+		if (this.supplier != null) {
+			return this.supplier.get();
+		}
+		throw new IllegalStateException("No function defined");
+	}
+
+	private static Class<?> getStartClass() {
+		ClassLoader classLoader = SpringFunctionInitializer.class.getClassLoader();
+		if (System.getenv("MAIN_CLASS") != null) {
+			return ClassUtils.resolveClassName(System.getenv("MAIN_CLASS"), classLoader);
+		}
+		try {
+			Class<?> result = getStartClass(
+					Collections.list(classLoader.getResources("META-INF/MANIFEST.MF")));
+			if (result == null) {
+				result = getStartClass(Collections
+						.list(classLoader.getResources("meta-inf/manifest.mf")));
+			}
+			logger.info("Main class: " + result);
+			return result;
+		}
+		catch (Exception ex) {
+			logger.error("Failed to find main class", ex);
+			return null;
+		}
+	}
+
+	private static Class<?> getStartClass(List<URL> list) {
+		logger.info("Searching manifests: " + list);
+		for (URL url : list) {
+			try {
+				logger.info("Searching manifest: " + url);
+				InputStream inputStream = url.openStream();
+				try {
+					Manifest manifest = new Manifest(inputStream);
+					String startClass = manifest.getMainAttributes()
+							.getValue("Start-Class");
+					if (startClass != null) {
+						return ClassUtils.forName(startClass,
+								SpringFunctionInitializer.class.getClassLoader());
+					}
+				}
+				finally {
+					inputStream.close();
+				}
+			}
+			catch (Exception ex) {
+			}
+		}
+		return null;
+	}
+
+}

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

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2016-2017 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.aws;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import org.junit.After;
+import org.junit.Test;
+
+import org.springframework.cloud.function.context.ContextFunctionCatalogAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class SpringFunctionInitializerTests {
+	
+	private SpringFunctionInitializer initializer;
+
+	@After
+	public void after() {
+		System.clearProperty("function.name");
+		if (initializer!=null) {
+			initializer.close();
+		}
+	}
+
+	@Test
+	public void functionBean() {
+		initializer = new SpringFunctionInitializer(
+				FluxFunctionConfig.class);
+		initializer.initialize();
+		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
+	}
+
+	@Test
+	public void functionCatalog() {
+		initializer = new SpringFunctionInitializer(
+				FunctionConfig.class);
+		initializer.initialize();
+		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
+	}
+
+	@Test
+	public void namedFunctionCatalog() {
+		initializer = new SpringFunctionInitializer(
+				NamedFunctionConfig.class);
+		System.setProperty("function.name", "other");
+		initializer.initialize();
+		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
+	}
+
+	@Test
+	public void consumerCatalog() {
+		initializer = new SpringFunctionInitializer(
+				ConsumerConfig.class);
+		initializer.initialize();
+		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		assertThat(result.toStream().collect(Collectors.toList())).isEmpty();
+	}
+
+	@Test
+	public void supplierCatalog() {
+		initializer = new SpringFunctionInitializer(
+				SupplierConfig.class);
+		initializer.initialize();
+		Flux<?> result = initializer.apply(Flux.empty());
+		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
+	}
+
+	@Configuration
+	protected static class FluxFunctionConfig {
+		@Bean
+		public Function<Flux<Foo>, Flux<Bar>> function() {
+			return flux -> flux.map(foo -> new Bar());
+		}
+	}
+
+	@Configuration
+	@Import(ContextFunctionCatalogAutoConfiguration.class)
+	protected static class FunctionConfig {
+		@Bean
+		public Function<Foo, Bar> function() {
+			return foo -> new Bar();
+		}
+	}
+
+	@Configuration
+	@Import(ContextFunctionCatalogAutoConfiguration.class)
+	protected static class NamedFunctionConfig {
+		@Bean
+		public Function<Foo, Bar> other() {
+			return foo -> new Bar();
+		}
+	}
+
+	@Configuration
+	@Import(ContextFunctionCatalogAutoConfiguration.class)
+	protected static class SupplierConfig {
+		@Bean
+		public Supplier<Bar> supplier() {
+			return () -> new Bar();
+		}
+	}
+
+	@Configuration
+	@Import(ContextFunctionCatalogAutoConfiguration.class)
+	protected static class ConsumerConfig {
+		@Bean
+		public Consumer<Foo> consumer() {
+			return foo -> {
+			};
+		}
+	}
+
+	protected static class Foo {
+	}
+
+	protected static class Bar {
+	}
+}