SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Migrate router-sink

Browse Source 
Resolves https://github.com/spring-cloud/stream-applications/issues/26
This commit is contained in:
Soby Chacko
2020-05-11 17:34:38 -04:00
parent 43e3d7a8fb
commit b2f2f81890
10 changed files with 614 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
applications/processor/groovy-processor/pom.xml
View File

@@ -88,7 +88,7 @@
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
<configuration>
<generatedApp>
<name>groovy-transform</name>
<name>groovy</name>
<type>processor</type>
<version>${project.version}</version>
<configClass>

11
applications/sink/pom.xml
View File

@@ -10,13 +10,14 @@
<packaging>pom</packaging>
<modules>
<module>file-sink</module>
<module>rabbit-sink</module>
<module>redis-sink</module>
<module>log-sink</module>
<module>cassandra-sink</module>
<module>mongodb-sink</module>
<module>counter-sink</module>
<module>file-sink</module>
<module>log-sink</module>
<module>mongodb-sink</module>
<module>jdbc-sink</module>
<module>redis-sink</module>
<module>rabbit-sink</module>
<module>router-sink</module>
</modules>
</project>

18
applications/sink/router-sink/README.adoc Normal file
View File

@@ -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[]

122
applications/sink/router-sink/pom.xml Normal file
View File

@@ -0,0 +1,122 @@
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>router-sink</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>router-sink</name>
<description>log sink apps</description>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<properties>
<apache-ivy.version>2.4.0</apache-ivy.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-groovy</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-json</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-nio</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-templates</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-xml</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-dateutil</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-datetime</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-jaxb</artifactId>
</dependency>
<dependency>
<groupId>org.apache.ivy</groupId>
<artifactId>ivy</artifactId>
<version>${apache-ivy.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>payload-converter-function</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud.stream.app.plugin</groupId>
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
<configuration>
<generatedApp>
<name>router</name>
<type>sink</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.stream.app.router.sink.RouterSinkConfiguration.class
</configClass>
<functionDefinition>byteArrayTextToString|routerSinkConsumer</functionDefinition>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>router-sink</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
</repositories>
</project>

103
applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkConfiguration.java Normal file
View File

@@ -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<Message<?>> 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<String, Object> 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);
}
}

159
applications/sink/router-sink/src/main/java/org/springframework/cloud/stream/app/router/sink/RouterSinkProperties.java Normal file
View File

@@ -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<Message<?>, 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;
}
}

1
applications/sink/router-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties Normal file
View File

@@ -0,0 +1 @@
configuration-properties.classes=org.springframework.cloud.fn.consumer.log.LogConsumerProperties

198
applications/sink/router-sink/src/test/java/org/springframework/cloud/stream/app/router/sink/RouterSinkIntegrationTests.java Normal file
View File

@@ -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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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 {
}
}

5
applications/sink/router-sink/src/test/resources/routertest.groovy Normal file
View File

@@ -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
}

1
applications/sink/router-sink/src/test/resources/routertest.properties Normal file
View File

@@ -0,0 +1 @@
bar=qux