diff --git a/applications/processor/filter-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/processor/filter-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index e0d1c859..523b1d70 100644 --- a/applications/processor/filter-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/processor/filter-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -1 +1 @@ -configuration-properties.classes=SpelFunctionProperties +configuration-properties.classes=org.springframework.cloud.fn.spel.SpelFunctionProperties diff --git a/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/filter/processor/FilterProcessorTests.java b/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/header/enricher/processor/FilterProcessorTests.java similarity index 97% rename from applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/filter/processor/FilterProcessorTests.java rename to applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/header/enricher/processor/FilterProcessorTests.java index bcfb11f8..2d52383d 100644 --- a/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/filter/processor/FilterProcessorTests.java +++ b/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/header/enricher/processor/FilterProcessorTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.app.filter.processor; +package org.springframework.cloud.stream.app.header.enricher.processor; import java.nio.charset.StandardCharsets; diff --git a/applications/processor/header-enricher-processor/README.adoc b/applications/processor/header-enricher-processor/README.adoc new file mode 100644 index 00000000..c9a934e3 --- /dev/null +++ b/applications/processor/header-enricher-processor/README.adoc @@ -0,0 +1,16 @@ +//tag::ref-doc[] += Header Enricher Processor + +Use the header-enricher app to add message headers. + +The headers are provided in the form of new line delimited key value pairs, where the keys are the header names and the values are SpEL expressions. +For example `--headers='foo=payload.someProperty \n bar=payload.otherProperty'`. + +== Options + +The **$$header-enricher$$** $$processor$$ has the following options: + +//tag::configuration-properties[] +//end::configuration-properties[] + +//end::ref-doc[] \ No newline at end of file diff --git a/applications/processor/header-enricher-processor/pom.xml b/applications/processor/header-enricher-processor/pom.xml new file mode 100644 index 00000000..a66b9540 --- /dev/null +++ b/applications/processor/header-enricher-processor/pom.xml @@ -0,0 +1,91 @@ + + + 4.0.0 + header-enricher-processor + header-enricher-processor + header-enricher processor apps + 3.0.0-SNAPSHOT + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.cloud.fn + header-enricher-function + ${java-functions.version} + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + org.springframework.integration + spring-integration-test-support + test + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + filter + processor + ${project.version} + org.springframework.cloud.fn.header.enricher.HeaderEnricherFunctionConfiguration.class + headerEnricherFunction + + + + + org.springframework.cloud.fn + header-enricher-function + ${java-functions.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/processor/header-enricher-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/processor/header-enricher-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..7e747b2e --- /dev/null +++ b/applications/processor/header-enricher-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.header.enricher.HeaderEnricherFunctionProperties + diff --git a/applications/processor/header-enricher-processor/src/test/java/org/springframework/cloud/stream/app/header/enricher/processor/HeaderEnricherProcessorTests.java b/applications/processor/header-enricher-processor/src/test/java/org/springframework/cloud/stream/app/header/enricher/processor/HeaderEnricherProcessorTests.java new file mode 100644 index 00000000..057e67d9 --- /dev/null +++ b/applications/processor/header-enricher-processor/src/test/java/org/springframework/cloud/stream/app/header/enricher/processor/HeaderEnricherProcessorTests.java @@ -0,0 +1,78 @@ +/* + * 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.header.enricher.processor; + +import org.hamcrest.MatcherAssert; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.fn.header.enricher.HeaderEnricherFunctionConfiguration; +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.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.integration.test.matcher.HeaderMatcher; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +import static org.hamcrest.Matchers.equalTo; + +/** + * @author Christian Tzolov + * @author Soby Chacko + */ +public class HeaderEnricherProcessorTests { + + @Test + public void testHeaderEnricherProcessor() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(HeaderEnricherProcessorConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=headerEnricherFunction", + "--header.enricher.headers=foo='bar' \n baz='fiz' \n buz=payload \n jaz=@value", + "--header.enricher.overwrite=true")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + OutputDestination processorOutput = context.getBean(OutputDestination.class); + + final Message message = MessageBuilder.withPayload("hello") + .setHeader("baz", "qux").build(); + processorInput.send(message); + Message enriched = processorOutput.receive(10000); + + MatcherAssert.assertThat(enriched, HeaderMatcher.hasHeader("foo", equalTo("bar"))); + MatcherAssert.assertThat(enriched, HeaderMatcher.hasHeader("baz", equalTo("fiz"))); + MatcherAssert.assertThat(enriched, HeaderMatcher.hasHeader("buz", equalTo("hello"))); + MatcherAssert.assertThat(enriched, HeaderMatcher.hasHeader("jaz", equalTo("beanValue"))); + } + } + + @EnableAutoConfiguration + @Import({HeaderEnricherFunctionConfiguration.class}) + public static class HeaderEnricherProcessorConfiguration { + + @Bean + public String value() { + return "beanValue"; + } + } + +} diff --git a/applications/processor/pom.xml b/applications/processor/pom.xml index 635a183d..8501a12c 100644 --- a/applications/processor/pom.xml +++ b/applications/processor/pom.xml @@ -11,6 +11,7 @@ bridge-processor + header-enricher-processor splitter-processor filter-processor transform-processor diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index 866eb793..679678d1 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -12,6 +12,7 @@ file-sink rabbit-sink + redis-sink log-sink cassandra-sink mongodb-sink diff --git a/applications/sink/redis-sink/README.adoc b/applications/sink/redis-sink/README.adoc new file mode 100644 index 00000000..bb5cb755 --- /dev/null +++ b/applications/sink/redis-sink/README.adoc @@ -0,0 +1,13 @@ +//tag::ref-doc[] += Redis Sink + +Sends messages to Redis. + +== Options + +The **$$redis$$** $$sink$$ has the following options: + +//tag::configuration-properties[] +//end::configuration-properties[] + +//end::ref-doc[] \ No newline at end of file diff --git a/applications/sink/redis-sink/pom.xml b/applications/sink/redis-sink/pom.xml new file mode 100644 index 00000000..5706a24c --- /dev/null +++ b/applications/sink/redis-sink/pom.xml @@ -0,0 +1,123 @@ + + + 4.0.0 + redis-sink + 3.0.0-SNAPSHOT + redis-sink + redis sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + 2.2.0.RELEASE + 1.48 + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + redis-consumer + ${java-functions.version} + + + + com.playtika.testcontainers + embedded-redis + ${embedded-redis.version} + test + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + org.springframework.cloud + spring-cloud-starter + ${spring-cloud-starters.version} + test + + + org.springframework.integration + spring-integration-test-support + test + + + org.awaitility + awaitility + test + + + junit + junit + + + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + redis + sink + ${project.version} + org.springframework.cloud.fn.consumer.redis.RedisConsumerConfiguration.class + + + + org.springframework.cloud.fn + redis-consumer + ${java-functions.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/redis-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/redis-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..afebf128 --- /dev/null +++ b/applications/sink/redis-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.redis.RedisConsumerProperties + diff --git a/applications/sink/redis-sink/src/test/java/org/springframework/cloud/stream/app/redis/sink/RedisSinkTests.java b/applications/sink/redis-sink/src/test/java/org/springframework/cloud/stream/app/redis/sink/RedisSinkTests.java new file mode 100644 index 00000000..e92f7f1d --- /dev/null +++ b/applications/sink/redis-sink/src/test/java/org/springframework/cloud/stream/app/redis/sink/RedisSinkTests.java @@ -0,0 +1,85 @@ +/* + * 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.redis.sink; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.fn.consumer.redis.RedisConsumerConfiguration; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Import; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.support.collections.DefaultRedisList; +import org.springframework.data.redis.support.collections.RedisList; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Soby Chacko + */ +public class RedisSinkTests { + + @Test + public void testRedisSink() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(RedisSinkConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=redisConsumer", + "--redis.consumer.key=foo")) { + + //Setup + String key = "foo"; + + final StringRedisTemplate redisTemplate = context.getBean(StringRedisTemplate.class); + redisTemplate.delete(key); + + RedisList redisList = new DefaultRedisList<>(key, redisTemplate); + List list = new ArrayList<>(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + + //Execute + Message> message = new GenericMessage<>(list); + + InputDestination source = context.getBean(InputDestination.class); + source.send(message); + + assertThat(redisList.size()).isEqualTo(3); + assertThat(redisList.get(0)).isEqualTo("Manny"); + assertThat(redisList.get(1)).isEqualTo("Moe"); + assertThat(redisList.get(2)).isEqualTo("Jack"); + } + } + + @EnableAutoConfiguration + @Import(RedisConsumerConfiguration.class) + public static class RedisSinkConfiguration { + } +} diff --git a/applications/sink/redis-sink/src/test/resources/application.properties b/applications/sink/redis-sink/src/test/resources/application.properties new file mode 100644 index 00000000..c286a3b2 --- /dev/null +++ b/applications/sink/redis-sink/src/test/resources/application.properties @@ -0,0 +1,3 @@ +spring.redis.host=${embedded.redis.host} +spring.redis.port=${embedded.redis.port} +spring.redis.password=${embedded.redis.password} \ No newline at end of file