Redis consumer

consumer/README.adoc

@@ -36,4 +36,8 @@ NOTE: A respective JDBC driver has to be added into classpath to make this consu
 
 `file-consumer`
 
-A `Consumer` that accepts incoming messages and write them into a target file in a directory.
+A `Consumer` that accepts incoming messages and write them into a target file in a directory.
+
+`redis-consumer`
+
+A `Consumer` that sends incoming messages to Redis.

consumer/redis-consumer/pom.xml

@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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-consumer</artifactId>
+    <version>1.0.0.BUILD-SNAPSHOT</version>
+    <name>redis-consumer</name>
+    <description>redis consumer</description>
+
+    <parent>
+        <groupId>io.pivotal.java.function</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0.BUILD-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</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-data-redis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-redis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.validation</groupId>
+            <artifactId>validation-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </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>
+    </dependencies>
+
+</project>

consumer/redis-consumer/src/main/java/io/pivotal/java/function/redis/consumer/RedisConsumerConfiguration.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2015-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.
+ */
+
+/**
+ * @author Eric Bottard
+ * @author Mark Pollack
+ * @author Gary Russell
+ * @author Soby Chacko
+ */
+package io.pivotal.java.function.redis.consumer;
+
+import java.util.function.Consumer;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
+import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
+import org.springframework.integration.redis.outbound.RedisStoreWritingMessageHandler;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+
+@Configuration
+@EnableConfigurationProperties(RedisConsumerProperties.class)
+public class RedisConsumerConfiguration {
+
+	private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
+
+	@Bean
+	public Consumer<Message<?>> redisConsumer(MessageHandler redisConsumerMessageHandler) {
+		return redisConsumerMessageHandler::handleMessage;
+	}
+
+	@Bean
+	public MessageHandler redisConsumerMessageHandler(RedisConnectionFactory redisConnectionFactory,
+													  RedisConsumerProperties redisConsumerProperties) {
+		if (redisConsumerProperties.isKeyPresent()) {
+			RedisStoreWritingMessageHandler redisStoreWritingMessageHandler = new RedisStoreWritingMessageHandler(
+					redisConnectionFactory);
+			redisStoreWritingMessageHandler.setKeyExpression(
+					new LiteralExpression(redisConsumerProperties.keyExpression()));
+			return redisStoreWritingMessageHandler;
+		}
+		else if (redisConsumerProperties.isQueuePresent()) {
+			return new RedisQueueOutboundChannelAdapter(redisConsumerProperties.queueExpression(),
+					redisConnectionFactory);
+		}
+		else { // must be topic
+			RedisPublishingMessageHandler redisPublishingMessageHandler = new RedisPublishingMessageHandler(
+					redisConnectionFactory);
+			redisPublishingMessageHandler.setTopicExpression(
+					EXPRESSION_PARSER.parseExpression(redisConsumerProperties.topicExpression()));
+			return redisPublishingMessageHandler;
+		}
+	}
+}

consumer/redis-consumer/src/main/java/io/pivotal/java/function/redis/consumer/RedisConsumerProperties.java

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2015-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 io.pivotal.java.function.redis.consumer;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import javax.validation.constraints.AssertTrue;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.util.StringUtils;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Used to configure those Redis Sink module options that are not related to connecting to Redis.
+ *
+ * @author Eric Bottard
+ * @author Mark Pollack
+ * @author Artem Bilan
+ * @author Soby Chacko
+ */
+@ConfigurationProperties("redis.consumer")
+@Validated
+public class RedisConsumerProperties {
+
+	/**
+	 * A SpEL expression to use for topic.
+	 */
+	private String topicExpression;
+
+	/**
+	 * A SpEL expression to use for queue.
+	 */
+	private String queueExpression;
+
+	/**
+	 * A SpEL expression to use for storing to a key.
+	 */
+	private String keyExpression;
+
+	/**
+	 * A literal key name to use when storing to a key.
+	 */
+	private String key;
+
+	/**
+	 * A literal queue name to use when storing in a queue.
+	 */
+	private String queue;
+
+	/**
+	 * A literal topic name to use when publishing to a topic.
+	 */
+	private String topic;
+
+	public void setKey(String key) {
+		this.key = key;
+	}
+
+	public void setQueue(String queue) {
+		this.queue = queue;
+	}
+
+	public void setTopic(String topic) {
+		this.topic = topic;
+	}
+
+	public String keyExpression() {
+		return key != null ? new LiteralExpression(key).getExpressionString() : keyExpression;
+	}
+
+	public void setKeyExpression(String keyExpression) {
+		this.keyExpression = keyExpression;
+	}
+
+	public String queueExpression() {
+		return queue != null ? new LiteralExpression(queue).getExpressionString() : queueExpression;
+	}
+
+	public void setQueueExpression(String queueExpression) {
+		this.queueExpression = queueExpression;
+	}
+
+	public String topicExpression() {
+		return topic != null ? new LiteralExpression(topic).getExpressionString() : topicExpression;
+	}
+
+	public void setTopicExpression(String topicExpression) {
+		this.topicExpression = topicExpression;
+	}
+
+	boolean isKeyPresent() {
+		return StringUtils.hasText(key) || keyExpression != null;
+	}
+
+	boolean isQueuePresent() {
+		return StringUtils.hasText(queue) || queueExpression != null;
+	}
+
+	boolean isTopicPresent() {
+		return StringUtils.hasText(topic) || topicExpression != null;
+	}
+
+	public String getTopicExpression() {
+		return topicExpression;
+	}
+
+	public String getQueueExpression() {
+		return queueExpression;
+	}
+
+	public String getKeyExpression() {
+		return keyExpression;
+	}
+
+	public String getKey() {
+		return key;
+	}
+
+	public String getQueue() {
+		return queue;
+	}
+
+	public String getTopic() {
+		return topic;
+	}
+
+
+	// The javabean property name is what will be reported in case of violation. Make it meaningful
+	@AssertTrue(message = "Exactly one of 'queue', 'queueExpression', 'key', 'keyExpression', "
+			+ "'topic' and 'topicExpression' must be set")
+	public boolean isMutuallyExclusive() {
+		Object[] props = new Object[] { queue, queueExpression, key, keyExpression, topic, topicExpression };
+		return (props.length - 1) == Collections.frequency(Arrays.asList(props), null);
+	}
+
+}