Add Header Enricher function and generate processor apps.

Add Redis consumer and gengerate sink apps.

Resolves https://github.com/spring-cloud/stream-applications/issues/22
Resolves https://github.com/spring-cloud/stream-applications/issues/23

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

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;
 

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

applications/processor/header-enricher-processor/pom.xml

@@ -0,0 +1,91 @@
+<?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>header-enricher-processor</artifactId>
+    <name>header-enricher-processor</name>
+    <description>header-enricher processor apps</description>
+    <version>3.0.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>org.springframework.cloud.stream.app</groupId>
+        <artifactId>stream-applications-core</artifactId>
+        <version>3.0.0-SNAPSHOT</version>
+        <relativePath/>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>header-enricher-function</artifactId>
+            <version>${java-functions.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test-support</artifactId>
+            <scope>test</scope>
+        </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>filter</name>
+                        <type>processor</type>
+                        <version>${project.version}</version>
+                        <configClass>org.springframework.cloud.fn.header.enricher.HeaderEnricherFunctionConfiguration.class</configClass>
+                        <functionDefinition>headerEnricherFunction</functionDefinition>
+                    </generatedApp>
+
+                    <dependencies>
+                        <dependency>
+                            <groupId>org.springframework.cloud.fn</groupId>
+                            <artifactId>header-enricher-function</artifactId>
+                            <version>${java-functions.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>

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
+

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<byte[]> 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";
+		}
+	}
+
+}

applications/processor/pom.xml

@@ -11,6 +11,7 @@
 
     <modules>
         <module>bridge-processor</module>
+        <module>header-enricher-processor</module>
         <module>splitter-processor</module>
         <module>filter-processor</module>
         <module>transform-processor</module>

applications/sink/pom.xml

@@ -12,6 +12,7 @@
     <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>

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

applications/sink/redis-sink/pom.xml

@@ -0,0 +1,123 @@
+<?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>redis-sink</artifactId>
+    <version>3.0.0-SNAPSHOT</version>
+    <name>redis-sink</name>
+    <description>redis 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>
+        <spring-cloud-starters.version>2.2.0.RELEASE</spring-cloud-starters.version>
+        <embedded-redis.version>1.48</embedded-redis.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>redis-consumer</artifactId>
+            <version>${java-functions.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.playtika.testcontainers</groupId>
+            <artifactId>embedded-redis</artifactId>
+            <version>${embedded-redis.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter</artifactId>
+            <version>${spring-cloud-starters.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test-support</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>junit</groupId>
+                    <artifactId>junit</artifactId>
+                </exclusion>
+            </exclusions>
+        </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>redis</name>
+                        <type>sink</type>
+                        <version>${project.version}</version>
+                        <configClass>org.springframework.cloud.fn.consumer.redis.RedisConsumerConfiguration.class</configClass>
+                    </generatedApp>
+                    <dependencies>
+                        <dependency>
+                            <groupId>org.springframework.cloud.fn</groupId>
+                            <artifactId>redis-consumer</artifactId>
+                            <version>${java-functions.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>

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
+

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<String> redisList = new DefaultRedisList<>(key, redisTemplate);
+			List<String> list = new ArrayList<>();
+			list.add("Manny");
+			list.add("Moe");
+			list.add("Jack");
+
+			//Execute
+			Message<List<String>> 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 {
+	}
+}

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}