diff --git a/README.adoc b/README.adoc index ebb809a2..f633a434 100644 --- a/README.adoc +++ b/README.adoc @@ -115,6 +115,60 @@ The following are the four major components of this repository. | |=== +=== Composite Function Support + +By default, the `source` applications are auto-configured with link:functions/function[functions] which may optionally be included in a composite function definition. +This feature enables the sources to do things like + - execute SpEL transformations + - enrich message headers + - filter events + - produce task launch requests + +or any combination of the above, without requiring a separate processor. + +For example, the time source executed, as shown below, will perform a series of transformations to publish a task launch request every second to the rabbit exchange `time-test`. + +``` +java -jar target/time-source-rabbit-3.0.0-SNAPSHOT.jar --spring.cloud.stream.bindings.output.destination=time-test --spring.cloud.stream.function.definition="timeSupplier|toMessage|spelFunction|headerEnricherFunction|taskLaunchRequestFunction" --spel.function.expression="payload.length()" --header.enricher.headers=task-id=payload*2 --task.launch.request.task-name-expression="'task-'+headers['task-id']" +``` + +The transformed message looks like this: + +``` +headers: +task-id: 34 +content_type: application/json +Payload +49 bytes +Encoding: string +{"args":[],"deploymentProps":{},"name":"task-34"} +``` + +Let's upack this function definition: + +`timeSupplier|toMessage|spelFunction|headerEnricherFunction|taskLaunchRequestFunction` + +This creates a composed Supplier beginning with the default `timeSupplier` which is the foundation for `time-source`. +This produces a String like `06/18/20 16:01:38`. We transform this using the SpEL expression `payload.length()`. +The `spelFunction` applies to a `Message` from which we can extract and transform the `payload` or `headers`, in accordance with common Spring Integration conventions. +We first need to wrap the date String in a message, using `toMessage`. + +NOTE: This step will be incorporated into a future release of Spring Cloud Function so may be omitted. + +The output of `spelFunction` is the length of the date-time String, `17`. +From here we apply the header enricher to add a Message header, `task-id` with the value of `payload*2` or ,`34`. +We use the `task-id` header to generate the task name for the task launch request using the SpEL expression "'task-'+headers['task-id']", or `task-34`. + +This somewhat contrived example is intended to show the power of function composition. +Even so, if you have `task-34` as a task definition in Data Flow, you could build a simple pipeline `time | tasklauncher` to launch the task every second. + +Prior to `3.0` release of Stream Applications, this composition required extensive customization. + +Please see the individual function documentation for an explanation of its configuration properties. + +NOTE: Support for composite functions includes auto-configuration for conventional binding name mappings (`input` and `output`) derived from the function definition and the presence of `spring.cloud.stream.bindings.output...`. +In this example, `--spring.cloud.stream.bindings.output.destination=time-test` is enabled behind the scenes by the auto-configured property +`--spring.cloud.stream.function.bindings.timeSuppliertoMessagespelFunctionheaderEnricherFunctiontaskLaunchRequestFunction-out-0=output`. === Build diff --git a/applications/source/file-source/pom.xml b/applications/source/file-source/pom.xml index 44c3adad..5f51b5cb 100644 --- a/applications/source/file-source/pom.xml +++ b/applications/source/file-source/pom.xml @@ -26,6 +26,12 @@ file-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + test + @@ -50,6 +56,11 @@ file-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/ftp-source/pom.xml b/applications/source/ftp-source/pom.xml index c5b97d81..44701003 100644 --- a/applications/source/ftp-source/pom.xml +++ b/applications/source/ftp-source/pom.xml @@ -30,6 +30,13 @@ org.springframework.cloud.fn function-test-support ${java-functions.version} + test + + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + test @@ -55,6 +62,11 @@ ftp-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java b/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java index f41d62e4..65a23dda 100644 --- a/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java +++ b/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java @@ -41,7 +41,8 @@ import static org.assertj.core.api.Assertions.assertThat; "ftp.factory.username = foo", "ftp.factory.password = foo", "file.consumer.mode = ref", - "ftp.factory.cacheSessions = true" + "ftp.factory.cacheSessions = true", + "spring.cloud.stream.function.definition=ftpSupplier" }) @DirtiesContext public class FtpSourceTests extends FtpTestSupport { diff --git a/applications/source/geode-source/pom.xml b/applications/source/geode-source/pom.xml index 9d19edcc..c4e0e106 100644 --- a/applications/source/geode-source/pom.xml +++ b/applications/source/geode-source/pom.xml @@ -67,6 +67,11 @@ geode-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/http-source/pom.xml b/applications/source/http-source/pom.xml index 58a7bd9e..ba9fe2d3 100644 --- a/applications/source/http-source/pom.xml +++ b/applications/source/http-source/pom.xml @@ -61,6 +61,11 @@ http-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/jdbc-source/pom.xml b/applications/source/jdbc-source/pom.xml index 7f03836c..400dabdc 100644 --- a/applications/source/jdbc-source/pom.xml +++ b/applications/source/jdbc-source/pom.xml @@ -72,6 +72,11 @@ jdbc-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/jms-source/pom.xml b/applications/source/jms-source/pom.xml index 66b4cb50..ce0e3503 100644 --- a/applications/source/jms-source/pom.xml +++ b/applications/source/jms-source/pom.xml @@ -60,6 +60,11 @@ jms-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/mongodb-source/pom.xml b/applications/source/mongodb-source/pom.xml index 1bc87f82..e2d677a7 100644 --- a/applications/source/mongodb-source/pom.xml +++ b/applications/source/mongodb-source/pom.xml @@ -52,6 +52,11 @@ mongodb-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/mqtt-source/pom.xml b/applications/source/mqtt-source/pom.xml index b1ed91e4..2ac48c02 100644 --- a/applications/source/mqtt-source/pom.xml +++ b/applications/source/mqtt-source/pom.xml @@ -56,6 +56,11 @@ mqtt-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/rabbit-source/pom.xml b/applications/source/rabbit-source/pom.xml index 500b9beb..8bc2367a 100644 --- a/applications/source/rabbit-source/pom.xml +++ b/applications/source/rabbit-source/pom.xml @@ -82,6 +82,11 @@ rabbit-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/tcp-source/pom.xml b/applications/source/tcp-source/pom.xml index f8dc1698..1158cec9 100644 --- a/applications/source/tcp-source/pom.xml +++ b/applications/source/tcp-source/pom.xml @@ -50,6 +50,11 @@ tcp-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/time-source/pom.xml b/applications/source/time-source/pom.xml index b9a8554d..38c76ff2 100644 --- a/applications/source/time-source/pom.xml +++ b/applications/source/time-source/pom.xml @@ -26,6 +26,12 @@ time-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + test + @@ -50,6 +56,11 @@ time-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java b/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java index 684c492a..b62361f8 100644 --- a/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java +++ b/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java @@ -16,9 +16,11 @@ package org.springframework.cloud.stream.app.source.time; +import java.io.IOException; import java.text.SimpleDateFormat; import java.util.Date; +import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; import org.springframework.boot.WebApplicationType; @@ -26,6 +28,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.fn.supplier.time.TimeSupplierConfiguration; import org.springframework.cloud.fn.supplier.time.TimeSupplierProperties; +import org.springframework.cloud.fn.task.launch.request.TaskLaunchRequest; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.ConfigurableApplicationContext; @@ -37,6 +40,7 @@ import static org.assertj.core.api.Assertions.assertThatCode; /** * @author Soby Chacko + * @author David Turanski */ public class TimeSourceTests { @@ -45,8 +49,8 @@ public class TimeSourceTests { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration .getCompleteConfiguration(TimeSourceTestApplication.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.function.definition=timeSupplier")) { + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=timeSupplier")) { OutputDestination target = context.getBean(OutputDestination.class); Message sourceMessage = target.receive(10000); @@ -61,6 +65,46 @@ public class TimeSourceTests { } } + @Test + public void testSourceComposedWithSpel() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(TimeSourceTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.definition=timeSupplier|toMessage|spelFunction", + "--spel.function.expression=payload.length()")) { + + OutputDestination target = context.getBean(OutputDestination.class); + Message sourceMessage = target.receive(10000); + final String actual = new String(sourceMessage.getPayload()); + assertThat(Integer.valueOf(actual)).isEqualTo(17); + } + } + + @Test + public void testSourceComposedWithOtherStuff() throws IOException { + ObjectMapper objectMapper = new ObjectMapper(); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(TimeSourceTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.definition=timeSupplier|toMessage|spelFunction|headerEnricherFunction|taskLaunchRequestFunction", + "--spel.function.expression=payload.length()", + "--header.enricher.headers=task-id=payload*2", + "--spring.cloud.stream.bindings.output.destination=foo", + "--task.launch.request.task-name-expression='task-'+headers['task-id']")) { + + OutputDestination target = context.getBean(OutputDestination.class); + Message sourceMessage = target.receive(10000); + TaskLaunchRequest taskLaunchRequest = objectMapper.readValue(sourceMessage.getPayload(), + TaskLaunchRequest.class); + assertThat(taskLaunchRequest.getTaskName()).isEqualTo("task-34"); + assertThat(context.getEnvironment().getProperty( + "spring.cloud.stream.function.bindings.timeSuppliertoMessagespelFunctionheaderEnricherFunctiontaskLaunchRequestFunction-out-0")) + .isEqualTo("output"); + } + } + @SpringBootApplication @Import(TimeSupplierConfiguration.class) public static class TimeSourceTestApplication { diff --git a/applications/source/websocket-source/pom.xml b/applications/source/websocket-source/pom.xml index 76e5d539..6a932f97 100644 --- a/applications/source/websocket-source/pom.xml +++ b/applications/source/websocket-source/pom.xml @@ -60,6 +60,11 @@ websocket-supplier ${java-functions.version} + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + diff --git a/applications/stream-applications-core/common/stream-applications-composite-function-support/pom.xml b/applications/stream-applications-core/common/stream-applications-composite-function-support/pom.xml new file mode 100644 index 00000000..1d7306cd --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-composite-function-support/pom.xml @@ -0,0 +1,70 @@ + + + + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + ../.. + + 4.0.0 + + stream-applications-composite-function-support + jar + stream-applications-composite-function-support + Common dependencies and configuration for composite functions + + + org.springframework.cloud.fn + filter-function + ${java-functions.version} + + + org.springframework.cloud.fn + spel-function + ${java-functions.version} + + + org.springframework.cloud.fn + header-enricher-function + ${java-functions.version} + + + org.springframework.cloud.fn + payload-converter-function + ${java-functions.version} + + + org.springframework.cloud.fn + splitter-function + ${java-functions.version} + + + org.springframework.cloud.fn + task-launch-request-function + ${java-functions.version} + + + org.springframework.boot + spring-boot-starter-test + test + + + diff --git a/applications/stream-applications-core/common/stream-applications-composite-function-support/src/main/java/org/springframework/cloud/stream/app/composite/function/common/FunctionBindingEnvironmentPostProcessor.java b/applications/stream-applications-core/common/stream-applications-composite-function-support/src/main/java/org/springframework/cloud/stream/app/composite/function/common/FunctionBindingEnvironmentPostProcessor.java new file mode 100644 index 00000000..7e945404 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-composite-function-support/src/main/java/org/springframework/cloud/stream/app/composite/function/common/FunctionBindingEnvironmentPostProcessor.java @@ -0,0 +1,110 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.stream.app.composite.function.common; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.context.properties.bind.BindResult; +import org.springframework.boot.context.properties.bind.Binder; +import org.springframework.boot.context.properties.source.ConfigurationPropertySources; +import org.springframework.boot.env.EnvironmentPostProcessor; +import org.springframework.core.env.ConfigurableEnvironment; +import org.springframework.core.env.Environment; +import org.springframework.core.env.MapPropertySource; +import org.springframework.util.StringUtils; + +/** + * Map default input and output to corresponding Spring Cloud Stream function binding + * channels for declared function name. + * @author David Turanski + */ + +public class FunctionBindingEnvironmentPostProcessor implements EnvironmentPostProcessor { + + private static Log log = LogFactory.getLog(FunctionBindingEnvironmentPostProcessor.class); + + private static final String DEFAULT_INPUT = "input"; + + private static final String DEFAULT_OUTPUT = "output"; + + private static final String OUT_0 = "-out-0"; + + private static final String IN_0 = "-in-0"; + + private static final String SPRING_CLOUD_STREAM_FUNCTION_BINDINGS_PREFIX = "spring.cloud.stream.function.bindings."; + + private static final String SPRING_CLOUD_STREAM_FUNCTION_DEFINITION = "spring.cloud.stream.function.definition"; + + private static final String SPRING_CLOUD_FUNCTION_DEFINITION = "spring.cloud.function.definition"; + + private static final String SPRING_CLOUD_STREAM_BINDINGS_PREFIX = "spring.cloud.stream.bindings"; + + @Override + public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) { + String functionDefinition = functionDefinition(environment); + if (StringUtils.isEmpty(functionDefinition)) { + return; + } + + Binder binder = new Binder(ConfigurationPropertySources.get(environment)); + BindResult bindResult = binder.bind(SPRING_CLOUD_STREAM_BINDINGS_PREFIX, HashMap.class); + if (bindResult.isBound()) { + Map functionBindings = new HashMap<>(); + + bindResult.get().keySet().forEach(bindingName -> { + if (bindingName.equals(DEFAULT_OUTPUT)) { + log.debug("Binding " + bindingName + " to output for function definition" + functionDefinition); + String key = functionBindingKeyName(functionDefinition, OUT_0); + functionBindings.put(key, bindingName); + } + if (bindingName.equals(DEFAULT_INPUT)) { + log.debug("Binding " + bindingName + " to function input for function definition" + + functionDefinition); + String key = functionBindingKeyName(functionDefinition, IN_0); + functionBindings.put(key, bindingName); + } + }); + if (!functionBindings.isEmpty()) { + environment.getPropertySources().addFirst(new MapPropertySource("function-bindings", functionBindings)); + } + } + } + + private String functionBindingKeyName(String functionDefinition, String suffix) { + return SPRING_CLOUD_STREAM_FUNCTION_BINDINGS_PREFIX + functionDefinitionToChannelName(functionDefinition) + + suffix; + } + + private String functionDefinitionToChannelName(String functionDefinition) { + return functionDefinition.replaceAll("\\|", ""); + } + + private String functionDefinition(Environment environment) { + if (environment.containsProperty(SPRING_CLOUD_STREAM_FUNCTION_DEFINITION)) { + return environment.getProperty(SPRING_CLOUD_STREAM_FUNCTION_DEFINITION); + } + else if (environment.containsProperty(SPRING_CLOUD_FUNCTION_DEFINITION)) { + return environment.getProperty(SPRING_CLOUD_FUNCTION_DEFINITION); + } + return null; + } +} diff --git a/applications/stream-applications-core/common/stream-applications-composite-function-support/src/main/resources/META-INF/spring.factories b/applications/stream-applications-core/common/stream-applications-composite-function-support/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..8b3cfeb4 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-composite-function-support/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.env.EnvironmentPostProcessor=\ + org.springframework.cloud.stream.app.composite.function.common.FunctionBindingEnvironmentPostProcessor diff --git a/applications/stream-applications-core/common/stream-applications-composite-function-support/src/test/java/org/springframework/cloud/stream/app/composite/function/common/FunctionBindingsEnvironmentPostProcessorTests.java b/applications/stream-applications-core/common/stream-applications-composite-function-support/src/test/java/org/springframework/cloud/stream/app/composite/function/common/FunctionBindingsEnvironmentPostProcessorTests.java new file mode 100644 index 00000000..89645b17 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-composite-function-support/src/test/java/org/springframework/cloud/stream/app/composite/function/common/FunctionBindingsEnvironmentPostProcessorTests.java @@ -0,0 +1,45 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.stream.app.composite.function.common; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ApplicationContext; + +import static org.assertj.core.api.Assertions.assertThat; + +public class FunctionBindingsEnvironmentPostProcessorTests { + + @Test + void destinationBindings() { + ApplicationContext context = new SpringApplication(TestApp.class).run( + "--spring.cloud.stream.bindings.output.destination=foo", + "--spring.cloud.stream.bindings.input.destination=bar", + "--spring.cloud.stream.function.definition=firstFunction|secondFunction"); + assertThat(context.getEnvironment().getProperty("spring.cloud.stream.function.bindings.firstFunctionsecondFunction-out-0")) + .isEqualTo("output"); + assertThat(context.getEnvironment().getProperty("spring.cloud.stream.function.bindings.firstFunctionsecondFunction-in-0")) + .isEqualTo("input"); + } + + @SpringBootApplication + static class TestApp { + + } +} diff --git a/applications/stream-applications-core/pom.xml b/applications/stream-applications-core/pom.xml index 6c4bc301..5b10107f 100644 --- a/applications/stream-applications-core/pom.xml +++ b/applications/stream-applications-core/pom.xml @@ -57,6 +57,7 @@ common/stream-applications-postprocessor-common common/stream-applications-micrometer-common common/stream-applications-security-common + common/stream-applications-composite-function-support diff --git a/functions/function/filter-function/src/main/resources/META-INF/spring.factories b/functions/function/filter-function/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..4823a19e --- /dev/null +++ b/functions/function/filter-function/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.filter.FilterFunctionConfiguration diff --git a/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionConfiguration.java b/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionConfiguration.java index afb025e3..85c8eedf 100644 --- a/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionConfiguration.java +++ b/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionConfiguration.java @@ -24,6 +24,7 @@ import java.util.function.Function; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -39,6 +40,7 @@ import org.springframework.messaging.Message; */ @Configuration @EnableConfigurationProperties(HeaderEnricherFunctionProperties.class) +@ConditionalOnProperty(prefix = "header.enricher", value = "headers") public class HeaderEnricherFunctionConfiguration { @Autowired diff --git a/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionProperties.java b/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionProperties.java index 2b24cf6a..d6681897 100644 --- a/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionProperties.java +++ b/functions/function/header-enricher-function/src/main/java/org/springframework/cloud/fn/header/enricher/HeaderEnricherFunctionProperties.java @@ -16,7 +16,6 @@ package org.springframework.cloud.fn.header.enricher; - import java.util.Properties; import javax.validation.constraints.NotNull; @@ -36,13 +35,13 @@ import org.springframework.validation.annotation.Validated; public class HeaderEnricherFunctionProperties { /** - * \n separated properties representing headers in which values are SpEL expressions, - * e.g foo='bar' \n baz=payload.baz. + * \n separated properties representing headers in which values are SpEL expressions, e.g + * foo='bar' \n baz=payload.baz. */ private Properties headers; /** - * set to true to overwrite any existing message headers. + * set to true to overwrite any existing message headers. */ private boolean overwrite = false; diff --git a/functions/function/header-enricher-function/src/main/resources/META-INF/spring.factories b/functions/function/header-enricher-function/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..15c09a96 --- /dev/null +++ b/functions/function/header-enricher-function/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.header.enricher.HeaderEnricherFunctionConfiguration diff --git a/functions/function/payload-converter-function/src/main/java/functions/ToMessage.java b/functions/function/payload-converter-function/src/main/java/functions/ToMessage.java new file mode 100644 index 00000000..f0d80a26 --- /dev/null +++ b/functions/function/payload-converter-function/src/main/java/functions/ToMessage.java @@ -0,0 +1,35 @@ +/* + * Copyright 2020-2020 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 + * + * https://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 functions; + +import java.util.function.Function; + +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +/** + * Wrap a POJO as a message. + * + */ +// TODO: Remove when https://github.com/spring-cloud/spring-cloud-function/issues/550 is +// resolved. +public class ToMessage implements Function> { + @Override + public Message apply(Object o) { + return MessageBuilder.withPayload(o).build(); + } +} diff --git a/functions/function/spel-function/src/main/resources/META-INF/spring.factories b/functions/function/spel-function/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..3c101209 --- /dev/null +++ b/functions/function/spel-function/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.spel.SpelFunctionConfiguration diff --git a/functions/function/splitter-function/src/main/resources/META-INF/spring.factories b/functions/function/splitter-function/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..cf51183f --- /dev/null +++ b/functions/function/splitter-function/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration diff --git a/functions/function/task-launch-request-function/src/main/resources/META-INF/spring.factories b/functions/function/task-launch-request-function/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..15653c2d --- /dev/null +++ b/functions/function/task-launch-request-function/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.task.launch.request.TaskLaunchRequestFunctionConfiguration