diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java index 1ec220432..e6cf5de87 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java @@ -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); diff --git a/spring-cloud-function-deployer-new/src/it/bootapp-multi/pom.xml b/spring-cloud-function-deployer-new/src/it/bootapp-multi/pom.xml new file mode 100644 index 000000000..07660f063 --- /dev/null +++ b/spring-cloud-function-deployer-new/src/it/bootapp-multi/pom.xml @@ -0,0 +1,71 @@ + + + 4.0.0 + + function.example + bootapp-multi + 0.0.1.BUILD-SNAPSHOT + jar + + + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + + + + + 1.8 + 3.0.0.BUILD-SNAPSHOT + 1.0.17.RELEASE + + + + + org.springframework.boot + spring-boot-starter + + + io.projectreactor + reactor-core + 3.1.2.RELEASE + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + org.apache.maven.plugins + maven-dependency-plugin + + + unpack + package + + unpack + + + + + ${project.groupId} + ${project.artifactId} + ${project.version} + exec + + + + + + + + + diff --git a/spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/MyFn.java b/spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/MyFn.java new file mode 100644 index 000000000..dd07b1d6d --- /dev/null +++ b/spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/MyFn.java @@ -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, Flux>, Tuple2, Flux>> { + + + @Override + public Tuple2, Flux> apply(Tuple2, Flux> inputs) { + Flux words = inputs.getT1(); + Flux numbers = inputs.getT2().publish().autoConnect(2); + + + Flux avg = numbers.buffer(2, 1) + .map(l -> l.stream().mapToInt(Integer::intValue).average().getAsDouble()) + .take(3); + + Flux repeated = words.zipWith(numbers) + .flatMap(t -> Flux.fromIterable(Collections.nCopies(t.getT2(), t.getT1()))); + + return Tuples.of(avg, repeated); + + } +} diff --git a/spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/RepeaterApplication.java b/spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/RepeaterApplication.java new file mode 100644 index 000000000..026a47a78 --- /dev/null +++ b/spring-cloud-function-deployer-new/src/it/bootapp-multi/src/main/java/function/example/RepeaterApplication.java @@ -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, Flux>, + Tuple2, Flux> + > fn() { + return new MyFn(); + } + + public static void main(String[] args) { + SpringApplication.run(RepeaterApplication.class, args); + } +} diff --git a/spring-cloud-function-deployer-new/src/it/bootjar-multi/pom.xml b/spring-cloud-function-deployer-new/src/it/bootjar-multi/pom.xml new file mode 100644 index 000000000..5a02732e7 --- /dev/null +++ b/spring-cloud-function-deployer-new/src/it/bootjar-multi/pom.xml @@ -0,0 +1,72 @@ + + + 4.0.0 + + function.example + bootjar-multi + 0.0.1.BUILD-SNAPSHOT + jar + + + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + + + + + 1.8 + 3.0.0.BUILD-SNAPSHOT + 1.0.17.RELEASE + + + + + org.springframework.boot + spring-boot-starter + + + io.projectreactor + reactor-core + 3.1.2.RELEASE + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + NONE + exec + + + + org.apache.maven.plugins + maven-dependency-plugin + + + unpack + package + + unpack + + + + + ${project.groupId} + ${project.artifactId} + ${project.version} + exec + + + + + + + + + diff --git a/spring-cloud-function-deployer-new/src/it/bootjar-multi/src/main/java/function/example/Repeater.java b/spring-cloud-function-deployer-new/src/it/bootjar-multi/src/main/java/function/example/Repeater.java new file mode 100644 index 000000000..643b9475d --- /dev/null +++ b/spring-cloud-function-deployer-new/src/it/bootjar-multi/src/main/java/function/example/Repeater.java @@ -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, Flux>, Tuple2, Flux>> { + + @Override + public Tuple2, Flux> apply(Tuple2, Flux> inputs) { + Flux stringFlux = inputs.getT1(); + Flux integerFlux = inputs.getT2(); + Flux sharedIntFlux = integerFlux.publish().autoConnect(2); + + Flux repeated = stringFlux.zipWith(sharedIntFlux) + .flatMap(t -> Flux.fromIterable(Collections.nCopies(t.getT2(), t.getT1()))); + + Flux sum = sharedIntFlux.buffer(2, 1) + .map(l -> l.stream().mapToInt(Integer::intValue).sum()) + ; + + return Tuples.of(repeated, sum); + } +} diff --git a/spring-cloud-function-deployer-new/src/main/java/org/springframework/cloud/function/deployer/FunctionArchiveDeployer.java b/spring-cloud-function-deployer-new/src/main/java/org/springframework/cloud/function/deployer/FunctionArchiveDeployer.java index 0c30fac3f..09cabfb27 100644 --- a/spring-cloud-function-deployer-new/src/main/java/org/springframework/cloud/function/deployer/FunctionArchiveDeployer.java +++ b/spring-cloud-function-deployer-new/src/main/java/org/springframework/cloud/function/deployer/FunctionArchiveDeployer.java @@ -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; diff --git a/spring-cloud-function-deployer-new/src/test/java/org/springframework/cloud/function/deployer/FunctionDeployerTests.java b/spring-cloud-function-deployer-new/src/test/java/org/springframework/cloud/function/deployer/FunctionDeployerTests.java index 2544b830f..94f532ace 100644 --- a/spring-cloud-function-deployer-new/src/test/java/org/springframework/cloud/function/deployer/FunctionDeployerTests.java +++ b/spring-cloud-function-deployer-new/src/test/java/org/springframework/cloud/function/deployer/FunctionDeployerTests.java @@ -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, Flux>, Tuple2, Flux>> + */ + @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>, Flux>>, Tuple2>, Flux>>> function = + catalog.lookup("fn", "application/json", "application/json"); + + Message msg1 = MessageBuilder.withPayload("\"one\"".getBytes()).build(); + Message msg2 = MessageBuilder.withPayload("\"two\"".getBytes()).build(); + Flux> inputOne = Flux.just(msg1, msg2); + + Message msgInt1 = MessageBuilder.withPayload("\"1\"".getBytes()).build(); + Message msgInt2 = MessageBuilder.withPayload("\"2\"".getBytes()).build(); + Flux> inputTwo = Flux.just(msgInt1, msgInt2); + + Tuple2>, Flux>> result = function.apply(Tuples.of(inputOne, inputTwo)); + List result1 = new ArrayList<>(); + List 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, Flux>, Tuple2, Flux>> + */ + @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>, Flux>>, Tuple2>, Flux>>> function = + catalog.lookup("repeater", "application/json", "application/json"); + + Message msg1 = MessageBuilder.withPayload("\"one\"".getBytes()).build(); + Message msg2 = MessageBuilder.withPayload("\"two\"".getBytes()).build(); + Flux> inputOne = Flux.just(msg1, msg2); + + Message msgInt1 = MessageBuilder.withPayload("\"1\"".getBytes()).build(); + Message msgInt2 = MessageBuilder.withPayload("\"2\"".getBytes()).build(); + Flux> inputTwo = Flux.just(msgInt1, msgInt2); + + Tuple2>, Flux>> result = function.apply(Tuples.of(inputOne, inputTwo)); + List result1 = new ArrayList<>(); + List 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"); + } + }