SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polished adapter initializers consolidation effort

Browse Source 
- Added ability to retrieve input type from FunctionRegistration (if available) in AbstractSpringFunctionAdapterInitializer
- Removed azure/AzureSpringFunctionInitializer and aws/SpringFunctionInitializer
- Added additional tests in AWS and Azure modules
- See 0189c578ef for additional info
This commit is contained in:
Oleg Zhurakousky
2019-02-25 19:37:34 +01:00
parent 41f66e9b48
commit cdca44f714
11 changed files with 102 additions and 558 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringBootKinesisEventHandler.java
View File

@@ -26,7 +26,6 @@ import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.messaging.Message; import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.GenericMessage;
@@ -37,6 +36,7 @@ import static java.util.stream.Collectors.toList;
* @param <O> response type * @param <O> response type
* @author Mark Fisher * @author Mark Fisher
* @author Halvdan Hoem Grelland * @author Halvdan Hoem Grelland
* @author Oleg Zhurakousky
*/ */
public class SpringBootKinesisEventHandler<E, O> public class SpringBootKinesisEventHandler<E, O>
extends SpringBootRequestHandler<KinesisEvent, O> { extends SpringBootRequestHandler<KinesisEvent, O> {
@@ -44,9 +44,6 @@ public class SpringBootKinesisEventHandler<E, O>
@Autowired @Autowired
private ObjectMapper mapper; private ObjectMapper mapper;
@Autowired
private FunctionInspector inspector;
public SpringBootKinesisEventHandler() { public SpringBootKinesisEventHandler() {
super(); super();
} }
@@ -65,7 +62,7 @@ public class SpringBootKinesisEventHandler<E, O>
protected Object convertEvent(KinesisEvent event) { protected Object convertEvent(KinesisEvent event) {
List<E> payloads = deserializePayloads(event.getRecords()); List<E> payloads = deserializePayloads(event.getRecords());
if (functionAcceptsMessage()) { if (getInspector().isMessage(function())) {
return wrapInMessages(payloads); return wrapInMessages(payloads);
} }
else { else {
@@ -92,9 +89,4 @@ public class SpringBootKinesisEventHandler<E, O>
throw new IllegalStateException("Cannot convert event", e); throw new IllegalStateException("Cannot convert event", e);
} }
} }
private boolean functionAcceptsMessage() {
return this.inspector.isMessage(function());
}
} }

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

@@ -25,12 +25,15 @@ import com.amazonaws.services.lambda.runtime.RequestHandler;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer;
/** /**
* @param <E> event type * @param <E> event type
* @param <O> result types * @param <O> result types
* @author Mark Fisher * @author Mark Fisher
* @author Oleg Zhurakousky
*/ */
public class SpringBootRequestHandler<E, O> extends SpringFunctionInitializer public class SpringBootRequestHandler<E, O> extends AbstractSpringFunctionAdapterInitializer<Context>
implements RequestHandler<E, Object> { implements RequestHandler<E, Object> {
public SpringBootRequestHandler(Class<?> configurationClass) { public SpringBootRequestHandler(Class<?> configurationClass) {
@@ -43,21 +46,31 @@ public class SpringBootRequestHandler<E, O> extends SpringFunctionInitializer
@Override @Override
public Object handleRequest(E event, Context context) { public Object handleRequest(E event, Context context) {
initialize(); initialize(context);
Object input = acceptsInput() ? convertEvent(event) : ""; Object input = acceptsInput() ? convertEvent(event) : "";
Publisher<?> output = apply(extract(input)); Publisher<?> output = apply(extract(input));
return result(input, output); return result(input, output);
} }
private Object result(Object input, Publisher<?> output) { @SuppressWarnings("unchecked")
@Override
protected <T> T result(Object input, Publisher<?> output) {
List<O> result = new ArrayList<>(); List<O> result = new ArrayList<>();
for (Object value : Flux.from(output).toIterable()) { for (Object value : Flux.from(output).toIterable()) {
result.add(convertOutput(value)); result.add(convertOutput(value));
} }
if (isSingleValue(input) && result.size() == 1) { if (isSingleValue(input) && result.size() == 1) {
return result.get(0); return (T) result.get(0);
} }
return result; return (T) result;
}
protected boolean acceptsInput() {
return !this.getInspector().getInputType(function()).equals(Void.class);
}
protected boolean returnsOutput() {
return !this.getInspector().getOutputType(function()).equals(Void.class);
} }
private boolean isSingleValue(Object input) { private boolean isSingleValue(Object input) {

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

@@ -49,14 +49,6 @@ public class SpringBootStreamHandler extends AbstractSpringFunctionAdapterInitia
super(configurationClass); super(configurationClass);
} }
@Override
protected void initialize(Context context) {
super.initialize(context);
if (this.mapper == null) {
this.mapper = new ObjectMapper();
}
}
@Override @Override
public void handleRequest(InputStream input, OutputStream output, Context context) public void handleRequest(InputStream input, OutputStream output, Context context)
throws IOException { throws IOException {
@@ -66,6 +58,15 @@ public class SpringBootStreamHandler extends AbstractSpringFunctionAdapterInitia
this.mapper.writeValue(output, result(value, flux)); this.mapper.writeValue(output, result(value, flux));
} }
@Override
protected void initialize(Context context) {
super.initialize(context);
if (this.mapper == null) {
this.mapper = new ObjectMapper();
}
}
private Flux<?> extract(Object input) { private Flux<?> extract(Object input) {
if (input instanceof Collection) { if (input instanceof Collection) {
return Flux.fromIterable((Iterable<?>) input); return Flux.fromIterable((Iterable<?>) input);

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

@@ -1,280 +0,0 @@
/*
* Copyright 2012-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.adapter.aws;
import java.io.Closeable;
import java.io.InputStream;
import java.net.URL;
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.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer;
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;
/**
* @author Dave Syer
* @author Semyon Fishman
* @deprecated as of 2.1 in favor of {@link AbstractSpringFunctionAdapterInitializer}.
* It is no longer used by the framework and only exists for avoiding potential regressions.
*/
@Deprecated
public class SpringFunctionInitializer implements Closeable {
private static Log logger = LogFactory.getLog(SpringFunctionInitializer.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 SpringFunctionInitializer(Class<?> configurationClass) {
this.configurationClass = configurationClass;
}
public SpringFunctionInitializer() {
this(getStartClass());
}
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;
}
@Override
public void close() {
if (this.context != null) {
this.context.close();
}
}
protected void initialize() {
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();
}
else {
initFunctionConsumerOrSupplierFromCatalog();
}
}
private String resolveName(Class<?> type) {
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);
}
@SuppressWarnings("unchecked")
private void initFunctionConsumerOrSupplierFromContext() {
String name = resolveName(Function.class);
if (context.containsBean(name) && context.getBean(name) instanceof Function) {
this.function = context.getBean(name, Function.class);
return;
}
name = resolveName(Consumer.class);
if (context.containsBean(name) && context.getBean(name) instanceof Consumer) {
this.consumer = context.getBean(name, Consumer.class);
return;
}
name = resolveName(Supplier.class);
if (context.containsBean(name) && context.getBean(name) instanceof Supplier) {
this.supplier = context.getBean(name, Supplier.class);
return;
}
}
private void initFunctionConsumerOrSupplierFromCatalog() {
String name = resolveName(Function.class);
this.function = this.catalog.lookup(Function.class, name);
if (this.function != null) {
return;
}
name = resolveName(Consumer.class);
this.consumer = this.catalog.lookup(Consumer.class, name);
if (this.consumer != null) {
return;
}
name = resolveName(Supplier.class);
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;
}
}
}
private SpringApplication springApplication() {
Class<?> sourceClass = this.configurationClass;
SpringApplication application = new org.springframework.cloud.function.context.FunctionalSpringApplication(
sourceClass);
application.setWebApplicationType(WebApplicationType.NONE);
return application;
}
protected Class<?> getInputType() {
if (this.inspector != null) {
return this.inspector.getInputType(function());
}
return Object.class;
}
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 boolean acceptsInput() {
return !this.inspector.getInputType(function()).equals(Void.class);
}
protected boolean returnsOutput() {
return !this.inspector.getOutputType(function()).equals(Void.class);
}
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");
}
}

7
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/test/java/org/springframework/cloud/function/adapter/aws/SpringBootApiGatewayRequestHandlerTests.java
View File

@@ -34,6 +34,9 @@ import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
/**
*
*/
public class SpringBootApiGatewayRequestHandlerTests { public class SpringBootApiGatewayRequestHandlerTests {
private SpringBootApiGatewayRequestHandler handler; private SpringBootApiGatewayRequestHandler handler;
@@ -41,8 +44,6 @@ public class SpringBootApiGatewayRequestHandlerTests {
@Test @Test
public void functionBean() { public void functionBean() {
this.handler = new SpringBootApiGatewayRequestHandler(FunctionConfig.class); this.handler = new SpringBootApiGatewayRequestHandler(FunctionConfig.class);
this.handler.initialize();
APIGatewayProxyRequestEvent request = new APIGatewayProxyRequestEvent(); APIGatewayProxyRequestEvent request = new APIGatewayProxyRequestEvent();
request.setBody("{\"value\":\"foo\"}"); request.setBody("{\"value\":\"foo\"}");
@@ -58,8 +59,6 @@ public class SpringBootApiGatewayRequestHandlerTests {
public void functionMessageBean() { public void functionMessageBean() {
this.handler = new SpringBootApiGatewayRequestHandler( this.handler = new SpringBootApiGatewayRequestHandler(
FunctionMessageConfig.class); FunctionMessageConfig.class);
this.handler.initialize();
APIGatewayProxyRequestEvent request = new APIGatewayProxyRequestEvent(); APIGatewayProxyRequestEvent request = new APIGatewayProxyRequestEvent();
request.setBody("{\"value\":\"foo\"}"); request.setBody("{\"value\":\"foo\"}");

3
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/test/java/org/springframework/cloud/function/adapter/aws/SpringBootKinesisEventHandlerTests.java
View File

@@ -111,7 +111,6 @@ public class SpringBootKinesisEventHandlerTests {
@Test @Test
public void functionBeanHandlesKinesisEvent() throws Exception { public void functionBeanHandlesKinesisEvent() throws Exception {
this.handler = new SpringBootKinesisEventHandler<>(FunctionConfig.class); this.handler = new SpringBootKinesisEventHandler<>(FunctionConfig.class);
this.handler.initialize();
KinesisEvent event = asKinesisEvent(singletonList(new Foo("foo"))); KinesisEvent event = asKinesisEvent(singletonList(new Foo("foo")));
@@ -123,7 +122,6 @@ public class SpringBootKinesisEventHandlerTests {
@Test @Test
public void functionBeanHandlesAggregatedKinesisEvent() throws Exception { public void functionBeanHandlesAggregatedKinesisEvent() throws Exception {
this.handler = new SpringBootKinesisEventHandler<>(FunctionConfig.class); this.handler = new SpringBootKinesisEventHandler<>(FunctionConfig.class);
this.handler.initialize();
List<Foo> events = asList(new Foo("foo"), new Foo("bar"), new Foo("baz")); List<Foo> events = asList(new Foo("foo"), new Foo("bar"), new Foo("baz"));
KinesisEvent aggregatedEvent = asAggregatedKinesisEvent(events); KinesisEvent aggregatedEvent = asAggregatedKinesisEvent(events);
@@ -137,7 +135,6 @@ public class SpringBootKinesisEventHandlerTests {
@Test @Test
public void functionMessageBean() throws Exception { public void functionMessageBean() throws Exception {
this.handler = new SpringBootKinesisEventHandler<>(FunctionMessageConfig.class); this.handler = new SpringBootKinesisEventHandler<>(FunctionMessageConfig.class);
this.handler.initialize();
KinesisEvent event = asKinesisEvent(asList(new Foo("foo"), new Foo("bar"))); KinesisEvent event = asKinesisEvent(asList(new Foo("foo"), new Foo("bar")));

1
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/test/java/org/springframework/cloud/function/adapter/aws/SpringBootRequestHandlerTests.java
View File

@@ -39,7 +39,6 @@ public class SpringBootRequestHandlerTests {
@Test @Test
public void functionBean() throws Exception { public void functionBean() throws Exception {
this.handler = new SpringBootRequestHandler<Foo, Bar>(FunctionConfig.class); this.handler = new SpringBootRequestHandler<Foo, Bar>(FunctionConfig.class);
this.handler.initialize();
Object output = this.handler.handleRequest(new Foo("foo"), null); Object output = this.handler.handleRequest(new Foo("foo"), null);
assertThat(output).isInstanceOf(Bar.class); assertThat(output).isInstanceOf(Bar.class);
} }

41
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/test/java/org/springframework/cloud/function/adapter/aws/SpringBootStreamHandlerTests.java
View File

@@ -21,6 +21,7 @@ import java.io.ByteArrayOutputStream;
import java.util.function.Function; import java.util.function.Function;
import org.junit.Test; import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration; import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration;
import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
@@ -58,6 +59,46 @@ public class SpringBootStreamHandlerTests {
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}"); assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
} }
@Test
public void functionNonFluxBeanNoCatalog() throws Exception {
this.handler = new SpringBootStreamHandler(NoCatalogNonFluxFunctionConfig.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Test
public void functionFluxBeanNoCatalog() throws Exception {
this.handler = new SpringBootStreamHandler(NoCatalogFluxFunctionConfig.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Configuration
protected static class NoCatalogNonFluxFunctionConfig {
@Bean
public Function<Foo, Bar> function() {
return foo -> new Bar(foo.getValue().toUpperCase());
}
}
@Configuration
protected static class NoCatalogFluxFunctionConfig {
@Bean
public Function<Flux<Foo>, Flux<Bar>> function() {
return flux -> flux.map(foo -> new Bar(foo.getValue().toUpperCase()));
}
}
@Configuration @Configuration
@Import({ ContextFunctionCatalogAutoConfiguration.class, @Import({ ContextFunctionCatalogAutoConfiguration.class,
JacksonAutoConfiguration.class }) JacksonAutoConfiguration.class })

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

@@ -1,243 +0,0 @@
/*
* Copyright 2012-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.adapter.azure;
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;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.jar.Manifest;
import com.microsoft.azure.functions.ExecutionContext;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer;
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;
/**
* @author Soby Chacko
* @deprecated as of 2.1 in favor of {@link AbstractSpringFunctionAdapterInitializer}.
* It is no longer used by the framework and only exists for avoiding potential regressions.
*/
@Deprecated
public class AzureSpringFunctionInitializer implements Closeable {
private volatile static ConfigurableApplicationContext context;
private final Class<?> configurationClass;
private volatile Function<Publisher<?>, Publisher<?>> function;
private AtomicBoolean initialized = new AtomicBoolean();
@Autowired(required = false)
private volatile FunctionCatalog catalog;
@Autowired(required = false)
private volatile FunctionInspector inspector;
public AzureSpringFunctionInitializer(Class<?> configurationClass) {
this.configurationClass = configurationClass == null ? getClass()
: configurationClass;
}
public AzureSpringFunctionInitializer() {
this(getStartClass());
}
private static Class<?> getStartClass() {
ClassLoader classLoader = org.springframework.cloud.function.adapter.azure.AzureSpringFunctionInitializer.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")));
}
return result;
}
catch (Exception ex) {
return null;
}
}
private static Class<?> getStartClass(List<URL> list) {
for (URL url : list) {
try {
InputStream inputStream = url.openStream();
try {
Manifest manifest = new Manifest(inputStream);
String startClass = manifest.getMainAttributes()
.getValue("Main-Class");
if (startClass != null) {
Class<?> aClass = ClassUtils.forName(startClass,
org.springframework.cloud.function.adapter.azure.AzureSpringFunctionInitializer.class
.getClassLoader());
SpringBootApplication declaredAnnotation = aClass
.getDeclaredAnnotation(SpringBootApplication.class);
if (declaredAnnotation != null) {
return aClass;
}
}
}
finally {
inputStream.close();
}
}
catch (Exception ex) {
}
}
return null;
}
@Override
public void close() throws IOException {
if (AzureSpringFunctionInitializer.context != null) {
AzureSpringFunctionInitializer.context.close();
AzureSpringFunctionInitializer.context = null;
}
}
@SuppressWarnings("unchecked")
protected void initialize(ExecutionContext ctxt) {
ConfigurableApplicationContext context = AzureSpringFunctionInitializer.context;
if (!this.initialized.compareAndSet(false, true)) {
return;
}
if (ctxt != null) {
ctxt.getLogger().info("Initializing functions");
}
if (context == null) {
synchronized (AzureSpringFunctionInitializer.class) {
if (context == null) {
ClassUtils.overrideThreadContextClassLoader(
AzureSpringFunctionInitializer.class.getClassLoader());
context = springApplication().run();
AzureSpringFunctionInitializer.context = context;
}
}
}
context.getAutowireCapableBeanFactory().autowireBean(this);
if (ctxt != null) {
ctxt.getLogger().info("Initialized context: catalog=" + this.catalog);
}
String name = context.getEnvironment().getProperty("function.name");
if (name == null) {
name = "function";
}
if (this.catalog == null) {
if (context.containsBean(name)) {
if (ctxt != null) {
ctxt.getLogger()
.info("No catalog. Looking for Function bean name=" + name);
}
this.function = context.getBean(name, Function.class);
}
}
else {
Set<String> functionNames = this.catalog.getNames(Function.class);
if (functionNames.size() == 1) {
this.function = this.catalog.lookup(Function.class,
functionNames.iterator().next());
}
else {
this.function = this.catalog.lookup(Function.class, name);
}
}
}
private SpringApplication springApplication() {
Class<?> sourceClass = this.configurationClass;
SpringApplication application = new org.springframework.cloud.function.context.FunctionalSpringApplication(
sourceClass);
application.setWebApplicationType(WebApplicationType.NONE);
return application;
}
protected boolean isSingleInput(Function<?, ?> function, Object input) {
if (!(input instanceof Collection)) {
return true;
}
if (this.inspector != null) {
return Collection.class
.isAssignableFrom(this.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(this.inspector.getOutputType(function));
}
return ((Collection<?>) output).size() <= 1;
}
protected Function<Publisher<?>, Publisher<?>> lookup(String name) {
Function<Publisher<?>, Publisher<?>> function = this.function;
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;
}
throw new IllegalStateException("No function defined with name=" + name);
}
public Function<Publisher<?>, Publisher<?>> getFunction() {
return this.function;
}
public FunctionCatalog getCatalog() {
return this.catalog;
}
public FunctionInspector getInspector() {
return this.inspector;
}
}

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

@@ -32,9 +32,10 @@ import org.springframework.context.annotation.Configuration;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
/** /**
* @author Dave Syer * @author Dave Syer
* * @author Oleg Zhurakousky
*/ */
public class AzureSpringBootRequestHandlerTests { public class AzureSpringBootRequestHandlerTests {
@@ -111,6 +112,13 @@ public class AzureSpringBootRequestHandlerTests {
assertThat(bar.getValue()).isEqualTo("BAR"); assertThat(bar.getValue()).isEqualTo("BAR");
} }
@Test
public void functionNonFluxBean() {
AzureSpringBootRequestHandler<Foo, Bar> handler = handler(NonFluxFunctionConfig.class);
Bar bar = handler.handleRequest(new Foo("bar"), new TestExecutionContext("function"));
assertThat(bar).isNotNull();
}
@After @After
public void close() throws IOException { public void close() throws IOException {
if (this.handler != null) { if (this.handler != null) {
@@ -118,6 +126,16 @@ public class AzureSpringBootRequestHandlerTests {
} }
} }
@Configuration
protected static class NonFluxFunctionConfig {
@Bean
public Function<Foo, Bar> function() {
return foo -> new Bar();
}
}
@Configuration @Configuration
protected static class BareConfig { protected static class BareConfig {

13
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/AbstractSpringFunctionAdapterInitializer.java
View File

@@ -65,6 +65,8 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
private Supplier<Publisher<?>> supplier; private Supplier<Publisher<?>> supplier;
private FunctionRegistration<?> functionRegistration;
private AtomicBoolean initialized = new AtomicBoolean(); private AtomicBoolean initialized = new AtomicBoolean();
@Autowired(required = false) @Autowired(required = false)
@@ -119,6 +121,9 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
if (this.inspector != null) { if (this.inspector != null) {
return this.inspector.getInputType(function()); return this.inspector.getInputType(function());
} }
else if (functionRegistration != null) {
return functionRegistration.getType().getInputType();
}
return Object.class; return Object.class;
} }
@@ -274,13 +279,15 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
@SuppressWarnings({ "unchecked", "rawtypes" }) @SuppressWarnings({ "unchecked", "rawtypes" })
private <T> T getAndInstrumentFromContext(String name) { private <T> T getAndInstrumentFromContext(String name) {
FunctionRegistration<?> functionRegistration = this.functionRegistration =
new FunctionRegistration(context.getBean(name), name); new FunctionRegistration(context.getBean(name), name);
Type type = FunctionContextUtils. Type type = FunctionContextUtils.
findType(name, (ConfigurableListableBeanFactory) this.context.getBeanFactory()); findType(name, (ConfigurableListableBeanFactory) this.context.getBeanFactory());
FunctionType functionType = new FunctionType(type);
return (T) functionRegistration.type(functionType).wrap().getTarget(); this.functionRegistration = functionRegistration.type(new FunctionType(type)).wrap();
return (T) functionRegistration.getTarget();
} }
private void initFunctionConsumerOrSupplierFromContext(Object targetContext) { private void initFunctionConsumerOrSupplierFromContext(Object targetContext) {