DATAREDIS-762 - Add AWS ElastiCache configuration.

We now provide RedisElastiCacheConfiguration to configure connections to AWS ElastiCache for Redis services with optional read replicas.

RedisElastiCacheConfiguration elastiCache = new RedisElastiCacheConfiguration("redis-node-1.elasticache.aws.com")
                                                   .node("redis-node-2.elasticache.aws.com");

LettuceConnectionFactory factory = new LettuceConnectionFactory(elastiCache, configuration);

Original Pull Request: #306

src/main/java/org/springframework/data/redis/connection/RedisElastiCacheConfiguration.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.redis.connection;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.util.Assert;
+
+/**
+ * Configuration class used for setting up {@link RedisConnection} via {@link RedisConnectionFactory} using connecting
+ * to <a href="https://aws.amazon.com/documentation/elasticache/">AWS ElastiCache with Read Replicas</a> .
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class RedisElastiCacheConfiguration {
+
+	private static final int DEFAULT_PORT = 6379;
+
+	private List<RedisStandaloneConfiguration> nodes = new ArrayList<>();
+	private int database;
+	private RedisPassword password = RedisPassword.none();
+
+	/**
+	 * Create a new {@link RedisElastiCacheConfiguration} given {@code hostName}.
+	 *
+	 * @param hostName must not be {@literal null} or empty.
+	 */
+	public RedisElastiCacheConfiguration(String hostName) {
+		this(hostName, DEFAULT_PORT);
+	}
+
+	/**
+	 * Create a new {@link RedisElastiCacheConfiguration} given {@code hostName} and {@code port}.
+	 *
+	 * @param hostName must not be {@literal null} or empty.
+	 * @param port a valid TCP port (1-65535).
+	 */
+	public RedisElastiCacheConfiguration(String hostName, int port) {
+		addNode(hostName, port);
+	}
+
+	/**
+	 * Add a {@link RedisStandaloneConfiguration node} to the list of nodes given {@code hostName}.
+	 *
+	 * @param hostName must not be {@literal null} or empty.
+	 * @param port a valid TCP port (1-65535).
+	 */
+	public void addNode(String hostName, int port) {
+		addNode(new RedisStandaloneConfiguration(hostName, port));
+	}
+
+	/**
+	 * Add a {@link RedisStandaloneConfiguration node} to the list of nodes.
+	 *
+	 * @param node must not be {@literal null}.
+	 */
+	private void addNode(RedisStandaloneConfiguration node) {
+
+		Assert.notNull(node, "RedisStandaloneConfiguration must not be null!");
+
+		node.setPassword(password);
+		node.setDatabase(database);
+		nodes.add(node);
+	}
+
+	/**
+	 * Add a {@link RedisStandaloneConfiguration node} to the list of nodes given {@code hostName}.
+	 *
+	 * @param hostName must not be {@literal null} or empty.
+	 * @return {@code this} {@link RedisElastiCacheConfiguration}.
+	 */
+	public RedisElastiCacheConfiguration node(String hostName) {
+		return node(hostName, DEFAULT_PORT);
+	}
+
+	/**
+	 * Add a {@link RedisStandaloneConfiguration node} to the list of nodes given {@code hostName} and {@code port}.
+	 *
+	 * @param hostName must not be {@literal null} or empty.
+	 * @param port a valid TCP port (1-65535).
+	 * @return {@code this} {@link RedisElastiCacheConfiguration}.
+	 */
+	public RedisElastiCacheConfiguration node(String hostName, int port) {
+
+		addNode(hostName, port);
+		return this;
+	}
+
+	/**
+	 * @return the db index.
+	 */
+	public int getDatabase() {
+		return database;
+	}
+
+	/**
+	 * Sets the index of the database used by this connection factory. Default is 0.
+	 *
+	 * @param index database index.
+	 */
+	public void setDatabase(int index) {
+
+		Assert.isTrue(index >= 0, () -> String.format("Invalid DB index '%s' (a positive index required)", index));
+
+		this.database = index;
+		this.nodes.forEach(it -> it.setDatabase(database));
+	}
+
+	/**
+	 * @return never {@literal null}.
+	 */
+	public RedisPassword getPassword() {
+		return password;
+	}
+
+	/**
+	 * @param password must not be {@literal null}.
+	 */
+	public void setPassword(RedisPassword password) {
+
+		Assert.notNull(password, "RedisPassword must not be null!");
+
+		this.password = password;
+		this.nodes.forEach(it -> it.setPassword(password));
+	}
+
+	/**
+	 * @return list of {@link RedisStandaloneConfiguration nodes}.
+	 */
+	public List<RedisStandaloneConfiguration> getNodes() {
+		return Collections.unmodifiableList(new ArrayList<>(nodes));
+	}
+}

src/main/java/org/springframework/data/redis/connection/lettuce/ElastiCacheConnectionProvider.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.redis.connection.lettuce;
+
+import io.lettuce.core.ReadFrom;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.StatefulConnection;
+import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.masterslave.MasterSlave;
+import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection;
+
+import java.util.Collection;
+import java.util.Optional;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * {@link LettuceConnectionProvider} implementation for a AWS ElastiCache with replicas setup. <br/>
+ * Lettuce auto-discovers node roles from the static {@link RedisURI} collection.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+class ElastiCacheConnectionProvider implements LettuceConnectionProvider {
+
+	private final RedisClient client;
+	private final RedisCodec<?, ?> codec;
+	private final Optional<ReadFrom> readFrom;
+	private final Collection<RedisURI> nodes;
+
+	/**
+	 * Create new {@link ElastiCacheConnectionProvider}.
+	 *
+	 * @param client must not be {@literal null}.
+	 * @param codec must not be {@literal null}.
+	 * @param nodes must not be {@literal null}.
+	 * @param readFrom can be {@literal null}.
+	 */
+	ElastiCacheConnectionProvider(RedisClient client, RedisCodec<?, ?> codec, Collection<RedisURI> nodes,
+			@Nullable ReadFrom readFrom) {
+
+		this.client = client;
+		this.codec = codec;
+		this.readFrom = Optional.ofNullable(readFrom);
+		this.nodes = nodes;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+	 */
+	@Override
+	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
+
+		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
+
+			StatefulRedisMasterSlaveConnection<?, ?> connection = MasterSlave.connect(client, codec, nodes);
+			readFrom.ifPresent(connection::setReadFrom);
+
+			return connectionType.cast(connection);
+		}
+
+		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
+	}
+}

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java

@@ -36,6 +36,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -68,6 +69,7 @@ import org.springframework.util.ClassUtils;
  * {@link LettuceConnectionFactory client configuration}. Lettuce supports the following environmental configurations:
  * <ul>
  * <li>{@link RedisStandaloneConfiguration}</li>
+ * <li>{@link RedisElastiCacheConfiguration}</li>
  * <li>{@link RedisSocketConfiguration}</li>
  * <li>{@link RedisSentinelConfiguration}</li>
  * <li>{@link RedisClusterConfiguration}</li>
@@ -101,6 +103,7 @@ public class LettuceConnectionFactory
 	private final Object connectionMonitor = new Object();
 	private boolean convertPipelineAndTxResults = true;
 	private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", 6379);
+	private @Nullable RedisElastiCacheConfiguration elastiCacheConfiguration;
 	private @Nullable RedisSocketConfiguration socketConfiguration;
 	private @Nullable RedisSentinelConfiguration sentinelConfiguration;
 	private @Nullable RedisClusterConfiguration clusterConfiguration;
@@ -199,6 +202,24 @@ public class LettuceConnectionFactory
 		this.standaloneConfig = standaloneConfig;
 	}
 
+	/**
+	 * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisElastiCacheConfiguration}
+	 * and {@link LettuceClientConfiguration}.
+	 *
+	 * @param elastiCacheConfiguration must not be {@literal null}.
+	 * @param clientConfig must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public LettuceConnectionFactory(RedisElastiCacheConfiguration elastiCacheConfiguration,
+			LettuceClientConfiguration clientConfig) {
+
+		this(clientConfig);
+
+		Assert.notNull(elastiCacheConfiguration, "RedisElastiCacheConfiguration must not be null!");
+
+		this.elastiCacheConfiguration = elastiCacheConfiguration;
+	}
+
 	/**
 	 * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSocketConfiguration} and
 	 * {@link LettuceClientConfiguration}.
@@ -625,6 +646,10 @@ public class LettuceConnectionFactory
 	 */
 	public int getDatabase() {
 
+		if (isElastiCacheAware()) {
+			return elastiCacheConfiguration.getDatabase();
+		}
+
 		if (isDomainSocketAware()) {
 			return socketConfiguration.getDatabase();
 		}
@@ -645,6 +670,11 @@ public class LettuceConnectionFactory
 
 		Assert.isTrue(index >= 0, "invalid DB index (a positive index required)");
 
+		if (isElastiCacheAware()) {
+			elastiCacheConfiguration.setDatabase(index);
+			return;
+		}
+
 		if (isDomainSocketAware()) {
 			socketConfiguration.setDatabase(index);
 			return;
@@ -694,6 +724,10 @@ public class LettuceConnectionFactory
 
 	private RedisPassword getRedisPassword() {
 
+		if (isElastiCacheAware()) {
+			return elastiCacheConfiguration.getPassword();
+		}
+
 		if (isDomainSocketAware()) {
 			return socketConfiguration.getPassword();
 		}
@@ -724,6 +758,11 @@ public class LettuceConnectionFactory
 			return;
 		}
 
+		if (isElastiCacheAware()) {
+			elastiCacheConfiguration.setPassword(RedisPassword.of(password));
+			return;
+		}
+
 		if (isRedisSentinelAware()) {
 			sentinelConfiguration.setPassword(RedisPassword.of(password));
 			return;
@@ -849,6 +888,14 @@ public class LettuceConnectionFactory
 		this.convertPipelineAndTxResults = convertPipelineAndTxResults;
 	}
 
+	/**
+	 * @return true when {@link RedisElastiCacheConfiguration} is present.
+	 * @since 2.1
+	 */
+	private boolean isElastiCacheAware() {
+		return elastiCacheConfiguration != null;
+	}
+
 	/**
 	 * @return true when {@link RedisSentinelConfiguration} is present.
 	 * @since 1.5
@@ -920,6 +967,16 @@ public class LettuceConnectionFactory
 
 		ReadFrom readFrom = getClientConfiguration().getReadFrom().orElse(null);
 
+		if (isElastiCacheAware()) {
+
+			List<RedisURI> nodes = this.elastiCacheConfiguration.getNodes().stream() //
+					.map(it -> createRedisURIAndApplySettings(it.getHostName(), it.getPort())) //
+					.peek(it -> it.setDatabase(getDatabase())) //
+					.collect(Collectors.toList());
+
+			return new ElastiCacheConnectionProvider((RedisClient) client, codec, nodes, readFrom);
+		}
+
 		if (isClusterAware()) {
 			return new ClusterConnectionProvider((RedisClusterClient) client, codec, readFrom);
 		}
@@ -929,6 +986,17 @@ public class LettuceConnectionFactory
 
 	protected AbstractRedisClient createClient() {
 
+		if (isElastiCacheAware()) {
+
+			RedisClient redisClient = clientConfiguration.getClientResources() //
+					.map(RedisClient::create) //
+					.orElseGet(RedisClient::create);
+
+			clientConfiguration.getClientOptions().ifPresent(redisClient::setOptions);
+
+			return redisClient;
+		}
+
 		if (isRedisSentinelAware()) {
 
 			RedisURI redisURI = getSentinelRedisURI();