diff --git a/applications/processor/groovy-processor/pom.xml b/applications/processor/groovy-processor/pom.xml index d300339d..6adfa06b 100644 --- a/applications/processor/groovy-processor/pom.xml +++ b/applications/processor/groovy-processor/pom.xml @@ -88,7 +88,7 @@ spring-cloud-stream-app-maven-plugin - groovy-transform + groovy processor ${project.version} diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index 679678d1..0d475725 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -10,13 +10,14 @@ pom - file-sink - rabbit-sink - redis-sink - log-sink cassandra-sink - mongodb-sink counter-sink + file-sink + log-sink + mongodb-sink jdbc-sink + redis-sink + rabbit-sink + router-sink diff --git a/applications/sink/router-sink/README.adoc b/applications/sink/router-sink/README.adoc new file mode 100644 index 00000000..5d85e5f0 --- /dev/null +++ b/applications/sink/router-sink/README.adoc @@ -0,0 +1,18 @@ +//tag::ref-doc[] += Log Sink + +The `log` sink uses the application logger to output the data for inspection. + +Please understand that `log` sink uses type-less handler, which affects how the actual logging will be performed. +This means that if the content-type is textual, then raw payload bytes will be converted to String, otherwise raw bytes will be logged. +Please see more info in the https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/#_content_type_versus_argument_type[user-guide]. + +== Options + +The **$$log$$** $$sink$$ has the following options: + + +//tag::configuration-properties[] +//end::configuration-properties[] + +//end::ref-doc[] diff --git a/applications/sink/router-sink/pom.xml b/applications/sink/router-sink/pom.xml new file mode 100644 index 00000000..8b54366f --- /dev/null +++ b/applications/sink/router-sink/pom.xml @@ -0,0 +1,122 @@ + + + 4.0.0 + router-sink + 3.0.0-SNAPSHOT + router-sink + log sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + 2.4.0 + + + + + org.springframework.integration + spring-integration-core + + + org.springframework.integration + spring-integration-groovy + + + org.codehaus.groovy + groovy-json + + + org.codehaus.groovy + groovy-nio + + + org.codehaus.groovy + groovy-templates + + + org.codehaus.groovy + groovy-xml + + + org.codehaus.groovy + groovy-dateutil + + + org.codehaus.groovy + groovy-datetime + + + org.codehaus.groovy + groovy-jaxb + + + org.apache.ivy + ivy + ${apache-ivy.version} + + + org.springframework.cloud.fn + payload-converter-function + 1.0.0-SNAPSHOT + + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + router + sink + ${project.version} + org.springframework.cloud.stream.app.router.sink.RouterSinkConfiguration.class + + byteArrayTextToString|routerSinkConsumer + + + + org.springframework.cloud.stream.app + router-sink + ${project.version} + + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkConfiguration.java b/applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkConfiguration.java new file mode 100644 index 00000000..38a18920 --- /dev/null +++ b/applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkConfiguration.java @@ -0,0 +1,103 @@ +/* + * Copyright 2016-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.router.sink; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.function.Consumer; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.support.PropertiesLoaderUtils; +import org.springframework.integration.groovy.GroovyScriptExecutingMessageProcessor; +import org.springframework.integration.router.AbstractMappingMessageRouter; +import org.springframework.integration.router.AbstractMessageRouter; +import org.springframework.integration.router.ExpressionEvaluatingRouter; +import org.springframework.integration.router.MessageRouter; +import org.springframework.integration.router.MethodInvokingRouter; +import org.springframework.integration.scripting.DefaultScriptVariableGenerator; +import org.springframework.integration.scripting.RefreshableResourceScriptSource; +import org.springframework.integration.scripting.ScriptVariableGenerator; +import org.springframework.messaging.Message; +import org.springframework.scripting.ScriptSource; +import org.springframework.util.CollectionUtils; + +/** + * A sink app that routes to one or more named channels. + * + * @author Gary Russell + * @author Artem Bilan + * @author Christian Tzolov + * @author Soby Chacko + */ +@Configuration +@EnableConfigurationProperties(RouterSinkProperties.class) +public class RouterSinkConfiguration { + + @Autowired + RouterSinkProperties properties; + + @Bean + public Consumer> routerSinkConsumer(MessageRouter router) { + return ((AbstractMessageRouter) router)::handleMessage; + } + + @Bean + public MessageRouter router(BinderAwareChannelResolver channelResolver, + ScriptVariableGenerator scriptVariableGenerator) { + AbstractMappingMessageRouter router; + if (properties.getScript() != null) { + router = new MethodInvokingRouter(scriptProcessor(scriptVariableGenerator, properties)); + } + else { + router = new ExpressionEvaluatingRouter(properties.getExpression()); + } + router.setDefaultOutputChannelName(properties.getDefaultOutputChannel()); + router.setResolutionRequired(properties.isResolutionRequired()); + if (properties.getDestinationMappings() != null) { + router.replaceChannelMappings(properties.getDestinationMappings()); + } + router.setChannelResolver(channelResolver); + return router; + } + + @Bean(name = "variableGenerator") + public ScriptVariableGenerator scriptVariableGenerator() throws IOException { + Map variables = new HashMap<>(); + CollectionUtils.mergePropertiesIntoMap(properties.getVariables(), variables); + if (properties.getVariablesLocation() != null) { + CollectionUtils.mergePropertiesIntoMap( + PropertiesLoaderUtils.loadProperties(properties.getVariablesLocation()), variables); + } + return new DefaultScriptVariableGenerator(variables); + } + + @Bean + @ConditionalOnProperty("router.script") + public GroovyScriptExecutingMessageProcessor scriptProcessor(ScriptVariableGenerator scriptVariableGenerator, + RouterSinkProperties properties) { + ScriptSource scriptSource = new RefreshableResourceScriptSource(properties.getScript(), + properties.getRefreshDelay()); + return new GroovyScriptExecutingMessageProcessor(scriptSource, scriptVariableGenerator); + } + +} diff --git a/applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkProperties.java b/applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkProperties.java new file mode 100644 index 00000000..a960330e --- /dev/null +++ b/applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkProperties.java @@ -0,0 +1,159 @@ +/* + * Copyright 2016-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.router.sink; + +import java.util.Properties; +import java.util.function.Function; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.core.io.Resource; +import org.springframework.expression.Expression; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.messaging.Message; + +/** + * Properties for the Router Sink; the router can use an expression + * or groovy script to return either a channel name, or a key to + * the channel mappings map. + * + * @author Gary Russell + */ +@ConfigurationProperties("router") +public class RouterSinkProperties { + + /** + * Default SpEL expression. + */ + public static final Expression DEFAULT_EXPRESSION = new FunctionExpression<>((Function, Object>) message -> message.getHeaders().get("routeTo")); + + /** + * Variable bindings as a new line delimited string of name-value pairs, e.g. 'foo=bar\n baz=car'. + */ + private Properties variables; + + /** + * The location of a properties file containing custom script variable bindings. + */ + private Resource variablesLocation; + + /** + * The expression to be applied to the message to determine the channel(s) to route to. + * Note that the payload wire format for content types such as text, json or xml is byte[] not String!. + * Consult the documentation for how to handle byte array payload content. + */ + private Expression expression = DEFAULT_EXPRESSION; + + /** + * The location of a groovy script that returns channels or channel mapping + * resolution keys. + */ + private Resource script; + + /** + * How often to check for script changes in ms (if present); < 0 means don't refresh. + */ + private int refreshDelay = 60000; + + /** + * Where to send un-routable messages. + */ + private String defaultOutputChannel = "nullChannel"; + + /** + * Whether or not channel resolution is required. + */ + private boolean resolutionRequired = false; + + /** + * Destination mappings as a new line delimited string of name-value pairs, e.g. 'foo=bar\n baz=car'. + */ + private Properties destinationMappings; + + public Properties getVariables() { + return variables; + } + + public void setVariables(Properties variables) { + this.variables = variables; + } + + public Resource getVariablesLocation() { + return variablesLocation; + } + + public void setVariablesLocation(Resource variablesLocation) { + this.variablesLocation = variablesLocation; + } + + public Expression getExpression() { + return this.expression; + } + + public void setExpression(Expression expression) { + this.expression = expression; + } + + public Resource getScript() { + return script; + } + + public void setScript(Resource script) { + this.script = script; + } + + @NotNull + public String getDefaultOutputChannel() { + return this.defaultOutputChannel; + } + + public void setDefaultOutputChannel(String defaultOutputChannel) { + this.defaultOutputChannel = defaultOutputChannel; + } + + public int getRefreshDelay() { + return refreshDelay; + } + + public void setRefreshDelay(int refreshDelay) { + this.refreshDelay = refreshDelay; + } + + public boolean isResolutionRequired() { + return this.resolutionRequired; + } + + public void setResolutionRequired(boolean resolutionRequired) { + this.resolutionRequired = resolutionRequired; + } + + public Properties getDestinationMappings() { + return destinationMappings; + } + + public void setDestinationMappings(Properties destinationMappings) { + this.destinationMappings = destinationMappings; + } + + @AssertTrue(message = "'expression' and 'script' are mutually exclusive") + public boolean isExpressionOrScriptValid() { + return this.script == null || this.expression == DEFAULT_EXPRESSION; + } + +} diff --git a/applications/sink/router-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/router-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..ab4592d7 --- /dev/null +++ b/applications/sink/router-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.log.LogConsumerProperties diff --git a/applications/sink/router-sink/src/test/java/org/springframework/cloud/stream/app/router/sink/RouterSinkIntegrationTests.java b/applications/sink/router-sink/src/test/java/org/springframework/cloud/stream/app/router/sink/RouterSinkIntegrationTests.java new file mode 100644 index 00000000..39998e74 --- /dev/null +++ b/applications/sink/router-sink/src/test/java/org/springframework/cloud/stream/app/router/sink/RouterSinkIntegrationTests.java @@ -0,0 +1,198 @@ +/* + * Copyright 2016-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.router.sink; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +import static org.assertj.core.api.Assertions.assertThat; + +public class RouterSinkIntegrationTests { + + @Test + public void testDefaultRouter() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(RouterSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=routerSinkConsumer", + "--router.resolutionRequired = true")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + + Message message = MessageBuilder.withPayload("hello").setHeader("routeTo", "baz").build(); + processorInput.send(message); + + OutputDestination processorOutput = context.getBean(OutputDestination.class); + Message sourceMessage = processorOutput.receive(10000); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("hello"); + } + } + + @Test + public void testDefaultRouterWithByteArrayPayload() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(RouterSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=routerSinkConsumer", + "--router.resolutionRequired = true")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + + Message message = MessageBuilder.withPayload("hello".getBytes()).setHeader("routeTo", "qux").build(); + processorInput.send(message); + + OutputDestination processorOutput = context.getBean(OutputDestination.class); + Message sourceMessage = processorOutput.receive(10000); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("hello"); + } + } + + @Test + public void testRouterWithExpression() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(RouterSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=routerSinkConsumer", + "--router.expression=headers['route']", + "--router.resolutionRequired=true")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + + Message message = MessageBuilder.withPayload("foo") + .setHeader("route", "foo").build(); + processorInput.send(message); + + OutputDestination processorOutput = context.getBean(OutputDestination.class); + Message sourceMessage = processorOutput.receive(10000); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("foo"); + } + } + + @Test + public void testRouterWithChannelMappings() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(RouterSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=routerSinkConsumer", + "--router.expression=headers['route']", + "--router.destinationMappings=foo=baz \n bar=qux", + "--router.resolutionRequired=true")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + + Message message = MessageBuilder.withPayload("foo") + .setHeader("route", "foo").build(); + processorInput.send(message); + + OutputDestination processorOutput = context.getBean(OutputDestination.class); + assertThat(context.getBean("baz")).isNotNull(); + + Message sourceMessage = processorOutput.receive(10000, "baz.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("foo"); + + message = MessageBuilder.withPayload("bar") + .setHeader("route", "bar").build(); + processorInput.send(message); + + processorOutput = context.getBean(OutputDestination.class); + assertThat(context.getBean("qux")).isNotNull(); + + sourceMessage = processorOutput.receive(10000, "qux.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("bar"); + } + } + + @Test + public void testWithDiscardedChannels() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(RouterSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=routerSinkConsumer", + "--router.expression=headers['route']", + "--router.defaultOutputChannel=discards", + "--spring.cloud.stream.dynamicDestinations=foo,bar,discards")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + + Message message = MessageBuilder.withPayload("foo") + .setHeader("route", "foo").build(); + processorInput.send(message); + + message = MessageBuilder.withPayload("bar") + .setHeader("route", "bar").build(); + processorInput.send(message); + + message = MessageBuilder.withPayload("hello") + .setHeader("route", "baz").build(); + processorInput.send(message); + + OutputDestination processorOutput = context.getBean(OutputDestination.class); + Message sourceMessage = processorOutput.receive(10000, "foo.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("foo"); + + sourceMessage = processorOutput.receive(10000, "bar.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("bar"); + + sourceMessage = processorOutput.receive(10000, "discards.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("hello"); + } + } + + @Test + public void testWithGroovyScript() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(RouterSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=routerSinkConsumer", + "--router.script=classpath:/routertest.groovy", + "--router.variables=foo=baz", + "--router.variablesLocation=classpath:/routertest.properties")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + + Message message = MessageBuilder.withPayload("foo") + .setHeader("route", "foo").build(); + processorInput.send(message); + + OutputDestination processorOutput = context.getBean(OutputDestination.class); + Message sourceMessage = processorOutput.receive(10000, "baz.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("foo"); + + + message = MessageBuilder.withPayload("bar") + .setHeader("route", "bar").build(); + processorInput.send(message); + + sourceMessage = processorOutput.receive(10000, "qux.destination"); + assertThat(new String(sourceMessage.getPayload())).isEqualTo("bar"); + } + } + + @SpringBootApplication + public static class RouterSinkTestApplication { + + } +} diff --git a/applications/sink/router-sink/src/test/resources/routertest.groovy b/applications/sink/router-sink/src/test/resources/routertest.groovy new file mode 100644 index 00000000..9db74b8a --- /dev/null +++ b/applications/sink/router-sink/src/test/resources/routertest.groovy @@ -0,0 +1,5 @@ +if (headers.route.equals('foo')) { + return "$foo" // mapped to baz in 'variables' +} else { + return "$bar" // mapped to qux in properties file +} diff --git a/applications/sink/router-sink/src/test/resources/routertest.properties b/applications/sink/router-sink/src/test/resources/routertest.properties new file mode 100644 index 00000000..1f2b6888 --- /dev/null +++ b/applications/sink/router-sink/src/test/resources/routertest.properties @@ -0,0 +1 @@ +bar=qux