GH-266 Consolidated Adapter infrastructure
- Moved common logic into a new AbstractSpringFunctionAdapterInitializer - Modified Azure and AWS request handlers to extend from it - Deprecated both AzureSpringFunctionInitializer and SpringFunctionInitializer(AWS) Resolves #266
This commit is contained in:
Oleg Zhurakousky
@@ -0,0 +1,370 @@
|
||||
/*
|
||||
* Copyright 2019-2019 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.context;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.InputStream;
|
||||
import java.lang.reflect.Type;
|
||||
import java.net.URL;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.Iterator;
|
||||
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.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionInspector;
|
||||
import org.springframework.cloud.function.context.config.FunctionContextUtils;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
/**
|
||||
* Base implementation for adapter initializers and request handlers.
|
||||
*
|
||||
* @param <C> the type of the target specific (native) context object.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.1
|
||||
*/
|
||||
public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Closeable {
|
||||
|
||||
private static Log logger = LogFactory.getLog(AbstractSpringFunctionAdapterInitializer.class);
|
||||
|
||||
private final Class<?> configurationClass;
|
||||
|
||||
private Function<Publisher<?>, Publisher<?>> function;
|
||||
|
||||
private Consumer<Publisher<?>> consumer;
|
||||
|
||||
private Supplier<Publisher<?>> supplier;
|
||||
|
||||
private AtomicBoolean initialized = new AtomicBoolean();
|
||||
|
||||
@Autowired(required = false)
|
||||
private FunctionInspector inspector;
|
||||
|
||||
@Autowired(required = false)
|
||||
private FunctionCatalog catalog;
|
||||
|
||||
private ConfigurableApplicationContext context;
|
||||
|
||||
public ConfigurableApplicationContext getContext() {
|
||||
return context;
|
||||
}
|
||||
|
||||
public AbstractSpringFunctionAdapterInitializer(Class<?> configurationClass) {
|
||||
this.configurationClass = configurationClass;
|
||||
}
|
||||
|
||||
public AbstractSpringFunctionAdapterInitializer() {
|
||||
this(getStartClass());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (this.context != null) {
|
||||
this.context.close();
|
||||
}
|
||||
}
|
||||
|
||||
protected void initialize(C targetContext) {
|
||||
if (!this.initialized.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
logger.info("Initializing: " + this.configurationClass);
|
||||
SpringApplication builder = springApplication();
|
||||
ConfigurableApplicationContext context = builder.run();
|
||||
context.getAutowireCapableBeanFactory().autowireBean(this);
|
||||
this.context = context;
|
||||
if (this.catalog == null) {
|
||||
initFunctionConsumerOrSupplierFromContext(targetContext);
|
||||
}
|
||||
else {
|
||||
initFunctionConsumerOrSupplierFromCatalog(targetContext);
|
||||
}
|
||||
}
|
||||
|
||||
protected FunctionInspector getInspector() {
|
||||
return inspector;
|
||||
}
|
||||
|
||||
protected Class<?> getInputType() {
|
||||
if (this.inspector != null) {
|
||||
return this.inspector.getInputType(function());
|
||||
}
|
||||
return Object.class;
|
||||
}
|
||||
|
||||
protected Function<Publisher<?>, Publisher<?>> getFunction() {
|
||||
return function;
|
||||
}
|
||||
|
||||
protected Object function() {
|
||||
if (this.function != null) {
|
||||
return this.function;
|
||||
}
|
||||
else if (this.consumer != null) {
|
||||
return this.consumer;
|
||||
}
|
||||
else if (this.supplier != null) {
|
||||
return this.supplier;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
protected Publisher<?> apply(Publisher<?> input) {
|
||||
if (this.function != null) {
|
||||
return Flux.from(this.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");
|
||||
}
|
||||
|
||||
/**
|
||||
* Allows you to resolve function name for cases where it
|
||||
* could not be located under default name.
|
||||
*
|
||||
* Default implementation returns empty string.
|
||||
*
|
||||
* @param targetContext the target context instance
|
||||
* @return the name of the function
|
||||
*/
|
||||
protected String doResolveName(Object targetContext) {
|
||||
return "";
|
||||
}
|
||||
|
||||
protected <O> O result(Object input, Publisher<?> output) {
|
||||
List<Object> result = new ArrayList<>();
|
||||
for (Object value : Flux.from(output).toIterable()) {
|
||||
result.add(value);
|
||||
}
|
||||
if (isSingleInput(getFunction(), input) && result.size() == 1) {
|
||||
@SuppressWarnings("unchecked")
|
||||
O value = (O) result.get(0);
|
||||
return value;
|
||||
}
|
||||
if (isSingleOutput(getFunction(), input) && result.size() == 1) {
|
||||
@SuppressWarnings("unchecked")
|
||||
O value = (O) result.get(0);
|
||||
return value;
|
||||
}
|
||||
@SuppressWarnings("unchecked")
|
||||
O value = (O) result;
|
||||
return value;
|
||||
}
|
||||
|
||||
private boolean isSingleInput(Function<?, ?> function, Object input) {
|
||||
if (!(input instanceof Collection)) {
|
||||
return true;
|
||||
}
|
||||
if (getInspector() != null) {
|
||||
return Collection.class
|
||||
.isAssignableFrom(getInspector().getInputType(function));
|
||||
}
|
||||
return ((Collection<?>) input).size() <= 1;
|
||||
}
|
||||
|
||||
private boolean isSingleOutput(Function<?, ?> function, Object output) {
|
||||
if (!(output instanceof Collection)) {
|
||||
return true;
|
||||
}
|
||||
if (getInspector() != null) {
|
||||
return Collection.class
|
||||
.isAssignableFrom(getInspector().getOutputType(function));
|
||||
}
|
||||
return ((Collection<?>) output).size() <= 1;
|
||||
}
|
||||
|
||||
private String resolveName(Class<?> type, Object targetContext) {
|
||||
String functionName = context.getEnvironment().getProperty("function.name");
|
||||
if (functionName != null) {
|
||||
return functionName;
|
||||
}
|
||||
else if (type.isAssignableFrom(Function.class)) {
|
||||
return "function";
|
||||
}
|
||||
else if (type.isAssignableFrom(Consumer.class)) {
|
||||
return "consumer";
|
||||
}
|
||||
else if (type.isAssignableFrom(Supplier.class)) {
|
||||
return "supplier";
|
||||
}
|
||||
throw new IllegalStateException("Unknown type " + type);
|
||||
}
|
||||
|
||||
private static Class<?> getStartClass() {
|
||||
ClassLoader classLoader = AbstractSpringFunctionAdapterInitializer.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,
|
||||
AbstractSpringFunctionAdapterInitializer.class.getClassLoader());
|
||||
}
|
||||
}
|
||||
finally {
|
||||
inputStream.close();
|
||||
}
|
||||
}
|
||||
catch (Exception ex) {
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
private <T> T getAndInstrumentFromContext(String name) {
|
||||
FunctionRegistration<?> functionRegistration =
|
||||
new FunctionRegistration(context.getBean(name), name);
|
||||
|
||||
Type type = FunctionContextUtils.
|
||||
findType(name, (ConfigurableListableBeanFactory) this.context.getBeanFactory());
|
||||
FunctionType functionType = new FunctionType(type);
|
||||
return (T) functionRegistration.type(functionType).wrap().getTarget();
|
||||
}
|
||||
|
||||
private void initFunctionConsumerOrSupplierFromContext(Object targetContext) {
|
||||
String name = resolveName(Function.class, targetContext);
|
||||
if (context.containsBean(name) && context.getBean(name) instanceof Function) {
|
||||
this.function = getAndInstrumentFromContext(name);
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Consumer.class, targetContext);
|
||||
if (context.containsBean(name) && context.getBean(name) instanceof Consumer) {
|
||||
this.consumer = getAndInstrumentFromContext(name);
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Supplier.class, targetContext);
|
||||
if (context.containsBean(name) && context.getBean(name) instanceof Supplier) {
|
||||
this.supplier = getAndInstrumentFromContext(name);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
private void initFunctionConsumerOrSupplierFromCatalog(Object targetContext) {
|
||||
String name = resolveName(Function.class, targetContext);
|
||||
this.function = this.catalog.lookup(Function.class, name);
|
||||
if (this.function != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Consumer.class, targetContext);
|
||||
this.consumer = this.catalog.lookup(Consumer.class, name);
|
||||
if (this.consumer != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Supplier.class, targetContext);
|
||||
this.supplier = this.catalog.lookup(Supplier.class, name);
|
||||
if (this.supplier != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.catalog.size() == 1) {
|
||||
Iterator<String> names = this.catalog.getNames(Function.class).iterator();
|
||||
if (names.hasNext()) {
|
||||
this.function = this.catalog.lookup(Function.class, names.next());
|
||||
return;
|
||||
}
|
||||
|
||||
names = this.catalog.getNames(Consumer.class).iterator();
|
||||
if (names.hasNext()) {
|
||||
this.consumer = this.catalog.lookup(Consumer.class, names.next());
|
||||
return;
|
||||
}
|
||||
|
||||
names = this.catalog.getNames(Supplier.class).iterator();
|
||||
if (names.hasNext()) {
|
||||
this.supplier = this.catalog.lookup(Supplier.class, names.next());
|
||||
return;
|
||||
}
|
||||
}
|
||||
else {
|
||||
name = this.doResolveName(targetContext);
|
||||
this.function = this.catalog.lookup(Function.class, name);
|
||||
if (this.function != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.consumer = this.catalog.lookup(Consumer.class, name);
|
||||
if (this.consumer != null) {
|
||||
return;
|
||||
}
|
||||
this.supplier = this.catalog.lookup(Supplier.class, name);
|
||||
if (this.supplier != null) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private SpringApplication springApplication() {
|
||||
Class<?> sourceClass = this.configurationClass;
|
||||
SpringApplication application = new org.springframework.cloud.function.context.FunctionalSpringApplication(
|
||||
sourceClass);
|
||||
application.setWebApplicationType(WebApplicationType.NONE);
|
||||
return application;
|
||||
}
|
||||
}
|
||||
@@ -40,7 +40,7 @@ import org.springframework.util.ReflectionUtils;
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.0
|
||||
*/
|
||||
abstract class FunctionContextUtils {
|
||||
public abstract class FunctionContextUtils {
|
||||
|
||||
public static Type findType(String name, ConfigurableListableBeanFactory registry) {
|
||||
AbstractBeanDefinition definition = (AbstractBeanDefinition) registry
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
/*
|
||||
* Copyright 2019-2019 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.context;
|
||||
|
||||
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.Ignore;
|
||||
import org.junit.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
|
||||
import org.springframework.context.ApplicationContextInitializer;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
|
||||
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
public class SpringFunctionAdapterInitializerTests {
|
||||
|
||||
private AbstractSpringFunctionAdapterInitializer<Object> initializer;
|
||||
|
||||
@After
|
||||
public void after() {
|
||||
System.clearProperty("function.name");
|
||||
if (this.initializer != null) {
|
||||
this.initializer.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void functionBean() {
|
||||
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FluxFunctionConfig.class) {
|
||||
|
||||
};
|
||||
this.initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
|
||||
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void functionApp() {
|
||||
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FluxFunctionApp.class) {
|
||||
|
||||
};
|
||||
this.initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
|
||||
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void functionCatalog() {
|
||||
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FunctionConfig.class) {
|
||||
|
||||
};
|
||||
this.initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
|
||||
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
@Ignore // related to boot 2.1 no bean override change
|
||||
public void functionRegistrar() {
|
||||
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FunctionRegistrar.class) {
|
||||
|
||||
};
|
||||
this.initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
|
||||
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void namedFunctionCatalog() {
|
||||
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(NamedFunctionConfig.class) {
|
||||
|
||||
};
|
||||
System.setProperty("function.name", "other");
|
||||
this.initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
|
||||
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void consumerCatalog() {
|
||||
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(ConsumerConfig.class) {
|
||||
|
||||
};
|
||||
this.initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
|
||||
assertThat(result.toStream().collect(Collectors.toList())).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void supplierCatalog() {
|
||||
initializer = new AbstractSpringFunctionAdapterInitializer<Object>(SupplierConfig.class) {
|
||||
|
||||
};
|
||||
initializer.initialize(null);
|
||||
Flux<?> result = Flux.from(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());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
protected static class FluxFunctionApp implements Function<Flux<Foo>, Flux<Bar>> {
|
||||
|
||||
@Override
|
||||
public Flux<Bar> apply(Flux<Foo> flux) {
|
||||
return flux.map(foo -> new Bar());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
protected static class FunctionRegistrar
|
||||
implements ApplicationContextInitializer<GenericApplicationContext> {
|
||||
|
||||
public Function<Flux<Foo>, Flux<Bar>> function() {
|
||||
return flux -> flux.map(foo -> new Bar());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void initialize(GenericApplicationContext context) {
|
||||
context.registerBean("function", FunctionRegistration.class,
|
||||
() -> new FunctionRegistration<Function<Flux<Foo>, Flux<Bar>>>(
|
||||
function()).name("function")
|
||||
.type(FunctionType.from(Foo.class).to(Bar.class)
|
||||
.wrap(Flux.class).getType()));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@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 {
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user