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

Improved archive loader to work with Java 11

Browse Source 
Added tests for deploying multi-input/output apps and jars
Polished logging
This commit is contained in:
Oleg Zhurakousky
2019-08-12 10:09:43 +02:00
parent 3349c3abed
commit b64af80fb1
8 changed files with 342 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java
View File

@@ -206,6 +206,9 @@ public class BeanFactoryAwareFunctionRegistry
@SuppressWarnings({ "unchecked", "rawtypes" })
private Function<?, ?> compose(Class<?> type, String definition, String... acceptedOutputTypes) {
if (logger.isInfoEnabled()) {
logger.info("Looking up function '" + definition + "' with acceptedOutputTypes: " + Arrays.asList(acceptedOutputTypes));
}
definition = discoverDefaultDefinitionIfNecessary(definition);
if (StringUtils.isEmpty(definition)) {
return null;
@@ -225,6 +228,8 @@ public class BeanFactoryAwareFunctionRegistry
for (String name : names) {
Object function = this.locateFunction(name);
if (function == null) {
logger.warn("!!! Failed to discover function '" + definition + "' in function catalog. "
+ "Function available in catalog are: " + this.getNames(null));
return null;
}
composedNameBuilder.append(prefix);

71
spring-cloud-function-deployer-new/src/it/bootapp-multi/pom.xml Normal file
View File

@@ -0,0 +1,71 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>function.example</groupId>
<artifactId>bootapp-multi</artifactId>
<version>0.0.1.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/>
</parent>
<properties>
<java.version>1.8</java.version>
<spring-cloud-function.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-function.version>
<wrapper.version>1.0.17.RELEASE</wrapper.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.1.2.RELEASE</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<classifier>exec</classifier>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>unpack</id>
<phase>package</phase>
<goals>
<goal>unpack</goal>
</goals>
<configuration>
<artifactItems>
<artifactItem>
<groupId>${project.groupId}</groupId>
<artifactId>${project.artifactId}</artifactId>
<version>${project.version}</version>
<classifier>exec</classifier>
</artifactItem>
</artifactItems>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

29
spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/MyFn.java Normal file
View File

@@ -0,0 +1,29 @@
package function.example;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
import java.util.Collections;
import java.util.function.Function;
public class MyFn implements Function<Tuple2<Flux<String>, Flux<Integer>>, Tuple2<Flux<Double>, Flux<String>>> {
@Override
public Tuple2<Flux<Double>, Flux<String>> apply(Tuple2<Flux<String>, Flux<Integer>> inputs) {
Flux<String> words = inputs.getT1();
Flux<Integer> numbers = inputs.getT2().publish().autoConnect(2);
Flux<Double> avg = numbers.buffer(2, 1)
.map(l -> l.stream().mapToInt(Integer::intValue).average().getAsDouble())
.take(3);
Flux<String> repeated = words.zipWith(numbers)
.flatMap(t -> Flux.fromIterable(Collections.nCopies(t.getT2(), t.getT1())));
return Tuples.of(avg, repeated);
}
}

24
spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/RepeaterApplication.java Normal file
View File

@@ -0,0 +1,24 @@
package function.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;
import java.util.function.Function;
@SpringBootApplication
public class RepeaterApplication {
@Bean
public Function<Tuple2<Flux<String>, Flux<Integer>>,
Tuple2<Flux<Double>, Flux<String>>
> fn() {
return new MyFn();
}
public static void main(String[] args) {
SpringApplication.run(RepeaterApplication.class, args);
}
}

72
spring-cloud-function-deployer-new/src/it/bootjar-multi/pom.xml Normal file
View File

@@ -0,0 +1,72 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>function.example</groupId>
<artifactId>bootjar-multi</artifactId>
<version>0.0.1.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/>
</parent>
<properties>
<java.version>1.8</java.version>
<spring-cloud-function.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-function.version>
<wrapper.version>1.0.17.RELEASE</wrapper.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.1.2.RELEASE</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<layout>NONE</layout>
<classifier>exec</classifier>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>unpack</id>
<phase>package</phase>
<goals>
<goal>unpack</goal>
</goals>
<configuration>
<artifactItems>
<artifactItem>
<groupId>${project.groupId}</groupId>
<artifactId>${project.artifactId}</artifactId>
<version>${project.version}</version>
<classifier>exec</classifier>
</artifactItem>
</artifactItems>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

27
spring-cloud-function-deployer-new/src/it/bootjar-multi/src/main/java/function/example/Repeater.java Normal file
View File

@@ -0,0 +1,27 @@
package function.example;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
import java.util.Collections;
import java.util.function.Function;
public class Repeater implements Function<Tuple2<Flux<String>, Flux<Integer>>, Tuple2<Flux<String>, Flux<Integer>>> {
@Override
public Tuple2<Flux<String>, Flux<Integer>> apply(Tuple2<Flux<String>, Flux<Integer>> inputs) {
Flux<String> stringFlux = inputs.getT1();
Flux<Integer> integerFlux = inputs.getT2();
Flux<Integer> sharedIntFlux = integerFlux.publish().autoConnect(2);
Flux<String> repeated = stringFlux.zipWith(sharedIntFlux)
.flatMap(t -> Flux.fromIterable(Collections.nCopies(t.getT2(), t.getT1())));
Flux<Integer> sum = sharedIntFlux.buffer(2, 1)
.map(l -> l.stream().mapToInt(Integer::intValue).sum())
;
return Tuples.of(repeated, sum);
}
}

44
spring-cloud-function-deployer-new/src/main/java/org/springframework/cloud/function/deployer/FunctionArchiveDeployer.java
View File

@@ -65,19 +65,8 @@ class FunctionArchiveDeployer extends JarLauncher {
super(archive);
}
void undeploy() {
this.stopDeployedApplicationContext();
try {
this.archiveLoader.close();
logger.info("Closed archive class loader");
}
catch (IOException e) {
logger.error("Failed to closed archive class loader", e);
}
}
@SuppressWarnings({ "unchecked", "rawtypes" })
protected void deploy(FunctionRegistry functionRegistry, FunctionProperties functionProperties, String[] args) {
void deploy(FunctionRegistry functionRegistry, FunctionProperties functionProperties, String[] args) {
ClassLoader currentLoader = Thread.currentThread().getContextClassLoader();
try {
@@ -118,6 +107,17 @@ class FunctionArchiveDeployer extends JarLauncher {
}
}
void undeploy() {
this.stopDeployedApplicationContext();
try {
this.archiveLoader.close();
logger.info("Closed archive class loader");
}
catch (IOException e) {
logger.error("Failed to closed archive class loader", e);
}
}
@Override
protected ClassLoader createClassLoader(URL[] urls) throws Exception {
String classAsPath = DeployerContextUtils.class.getName().replace('.', '/') + ".class";
@@ -132,7 +132,7 @@ class FunctionArchiveDeployer extends JarLauncher {
@Override
public Class<?> loadClass(String name) throws ClassNotFoundException {
Class<?> clazz = null;
if (name.startsWith("org.reactivestreams") || name.startsWith("reactor.")) {
if (shouldLoadViaDeployerLoader(name)) {
clazz = getClass().getClassLoader().loadClass(name);
}
else if (name.equals(DeployerContextUtils.class.getName())) {
@@ -157,7 +157,14 @@ class FunctionArchiveDeployer extends JarLauncher {
return this.archiveLoader;
}
protected boolean isBootApplicationWithMain() {
private boolean shouldLoadViaDeployerLoader(String name) {
return name.startsWith("org.reactivestreams")
|| name.startsWith("reactor.")
|| name.startsWith("java")
|| name.startsWith("com.sun");
}
private boolean isBootApplicationWithMain() {
try {
return StringUtils.hasText(this.getArchive().getManifest().getMainAttributes().getValue("Start-Class"));
}
@@ -187,9 +194,12 @@ class FunctionArchiveDeployer extends JarLauncher {
if (typeRef.get() != null) {
Object functionInstance = functionClass.newInstance();
functionRegistration = new FunctionRegistration<>(functionInstance,
StringUtils.uncapitalize(functionClass.getSimpleName()));
String functionName = StringUtils.uncapitalize(functionClass.getSimpleName());
if (logger.isInfoEnabled()) {
logger.info("Registering function class '" + functionClass + "' of type '" + typeRef.get()
+ "' under name '" + functionName + "'.");
}
functionRegistration = new FunctionRegistration<>(functionInstance, functionName);
functionRegistration.type(typeRef.get());
}
return functionRegistration;

87
spring-cloud-function-deployer-new/src/test/java/org/springframework/cloud/function/deployer/FunctionDeployerTests.java
View File

@@ -17,11 +17,14 @@
package org.springframework.cloud.function.deployer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
import org.springframework.boot.SpringApplication;
import org.springframework.cloud.function.context.FunctionCatalog;
@@ -125,4 +128,88 @@ public class FunctionDeployerTests {
assertThat(new String(result.getPayload(), StandardCharsets.UTF_8)).isEqualTo("{\"name\":\"BOB\",\"id\":1}");
}
/*
* Target Function
*
* @Bean Function<Tuple2<Flux<String>, Flux<Integer>>, Tuple2<Flux<Double>, Flux<String>>>
*/
@Test
public void testBootAppWithMultipleInputOutput() {
String[] args = new String[] {
"--spring.cloud.function.location=target/it/bootapp-multi/target/bootapp-multi-0.0.1.BUILD-SNAPSHOT-exec.jar",
"--spring.cloud.function.function-name=fn"
};
ApplicationContext context = SpringApplication.run(FunctionDeployerConfiguration.class, args);
FunctionCatalog catalog = context.getBean(FunctionCatalog.class);
Function<Tuple2<Flux<Message<byte[]>>, Flux<Message<byte[]>>>, Tuple2<Flux<Message<byte[]>>, Flux<Message<byte[]>>>> function =
catalog.lookup("fn", "application/json", "application/json");
Message<byte[]> msg1 = MessageBuilder.withPayload("\"one\"".getBytes()).build();
Message<byte[]> msg2 = MessageBuilder.withPayload("\"two\"".getBytes()).build();
Flux<Message<byte[]>> inputOne = Flux.just(msg1, msg2);
Message<byte[]> msgInt1 = MessageBuilder.withPayload("\"1\"".getBytes()).build();
Message<byte[]> msgInt2 = MessageBuilder.withPayload("\"2\"".getBytes()).build();
Flux<Message<byte[]>> inputTwo = Flux.just(msgInt1, msgInt2);
Tuple2<Flux<Message<byte[]>>, Flux<Message<byte[]>>> result = function.apply(Tuples.of(inputOne, inputTwo));
List<String> result1 = new ArrayList<>();
List<String> result2 = new ArrayList<>();
result.getT1().subscribe(message -> {
result1.add(new String(message.getPayload()));
});
result.getT2().subscribe(message -> {
result2.add(new String(message.getPayload()));
});
assertThat(result1.get(0)).isEqualTo("1.5");
assertThat(result1.get(1)).isEqualTo("2.0");
assertThat(result2.get(0)).isEqualTo("\"one\"");
assertThat(result2.get(1)).isEqualTo("\"two\"");
}
/*
* Target Function
*
* Function<Tuple2<Flux<String>, Flux<Integer>>, Tuple2<Flux<Double>, Flux<String>>>
*/
@Test
public void testBootJarWithMultipleInputOutput() {
String[] args = new String[] {
"--spring.cloud.function.location=target/it/bootjar-multi/target/bootjar-multi-0.0.1.BUILD-SNAPSHOT-exec.jar",
"--spring.cloud.function.function-class=function.example.Repeater"
};
ApplicationContext context = SpringApplication.run(FunctionDeployerConfiguration.class, args);
FunctionCatalog catalog = context.getBean(FunctionCatalog.class);
Function<Tuple2<Flux<Message<byte[]>>, Flux<Message<byte[]>>>, Tuple2<Flux<Message<byte[]>>, Flux<Message<byte[]>>>> function =
catalog.lookup("repeater", "application/json", "application/json");
Message<byte[]> msg1 = MessageBuilder.withPayload("\"one\"".getBytes()).build();
Message<byte[]> msg2 = MessageBuilder.withPayload("\"two\"".getBytes()).build();
Flux<Message<byte[]>> inputOne = Flux.just(msg1, msg2);
Message<byte[]> msgInt1 = MessageBuilder.withPayload("\"1\"".getBytes()).build();
Message<byte[]> msgInt2 = MessageBuilder.withPayload("\"2\"".getBytes()).build();
Flux<Message<byte[]>> inputTwo = Flux.just(msgInt1, msgInt2);
Tuple2<Flux<Message<byte[]>>, Flux<Message<byte[]>>> result = function.apply(Tuples.of(inputOne, inputTwo));
List<String> result1 = new ArrayList<>();
List<String> result2 = new ArrayList<>();
result.getT1().subscribe(message -> {
result1.add(new String(message.getPayload()));
});
result.getT2().subscribe(message -> {
result2.add(new String(message.getPayload()));
});
assertThat(result1.get(0)).isEqualTo("\"one\"");
assertThat(result1.get(1)).isEqualTo("\"two\"");
assertThat(result2.get(0)).isEqualTo("3");
assertThat(result2.get(1)).isEqualTo("2");
}
}