DATAREDIS-580 - Support Master/Slave connections with ReadFrom settings using Lettuce.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-redis</artifactId>
-	<version>2.1.0.BUILD-SNAPSHOT</version>
+	<version>2.1.0.DATAREDIS-580-SNAPSHOT</version>
 
 	<name>Spring Data Redis</name>
 

src/main/asciidoc/new-features.adoc

@@ -7,6 +7,7 @@ New and noteworthy in the latest releases.
 == New in Spring Data Redis 2.1
 
 * Unix domain socket connections using <<redis:connectors:lettuce,Lettuce>>.
+* <<redis:write-to-master-read-from-slave, Write to Master read from Slave>> support using Lettuce.
 
 [[new-in-2.0.0]]
 == New in Spring Data Redis 2.0

src/main/asciidoc/reference/redis.adoc

@@ -111,6 +111,30 @@ class RedisConfiguration {
 }
 ----
 
+[[redis:write-to-master-read-from-slave]]
+=== Write to Master read from Slave
+
+Redis Master/Slave setup, without automatic failover (for automatic failover see: <<redis:sentinel, Sentinel>>), not only allows data to be savely stored at more nodes. It also allows, using <<redis:connectors:lettuce, Lettuce>>,  reading data from slaves while pushing writes to the master.
+Set the read/write strategy to be used via `LettuceClientConfiguration`.
+
+[source,java]
+----
+@Configuration
+class WriteToMasterReadFromSlaveConfiguration {
+
+  @Bean
+  public LettuceConnectionFactory redisConnectionFactory() {
+
+    LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
+      .readFrom(SLAVE_PREFERRED)
+      .build();
+
+    return new LettuceConnectionFactory(new RedisStandaloneConfiguration("server", 6379), clientConfig);
+  }
+}
+----
+
+
 [[redis:sentinel]]
 == Redis Sentinel Support
 

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

@@ -15,12 +15,18 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import io.lettuce.core.ReadFrom;
 import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 
+import java.util.Optional;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
 /**
  * Connection provider for Cluster connections.
  *
@@ -32,11 +38,34 @@ class ClusterConnectionProvider implements LettuceConnectionProvider {
 
 	private final RedisClusterClient client;
 	private final RedisCodec<?, ?> codec;
+	private final Optional<ReadFrom> readFrom;
 
+	/**
+	 * Create new {@link ClusterConnectionProvider}.
+	 *
+	 * @param client must not be {@literal null}.
+	 * @param codec must not be {@literal null}.
+	 */
 	ClusterConnectionProvider(RedisClusterClient client, RedisCodec<?, ?> codec) {
+		this(client, codec, null);
+	}
+
+	/**
+	 * Create new {@link ClusterConnectionProvider}.
+	 *
+	 * @param client must not be {@literal null}.
+	 * @param codec must not be {@literal null}.
+	 * @param readFrom can be {@literal null}.
+	 * @since 2.1
+	 */
+	ClusterConnectionProvider(RedisClusterClient client, RedisCodec<?, ?> codec, @Nullable ReadFrom readFrom) {
+
+		Assert.notNull(client, "Client must not be null!");
+		Assert.notNull(codec, "Codec must not be null!");
 
 		this.client = client;
 		this.codec = codec;
+		this.readFrom = Optional.ofNullable(readFrom);
 	}
 
 	/*
@@ -52,7 +81,11 @@ class ClusterConnectionProvider implements LettuceConnectionProvider {
 
 		if (StatefulRedisClusterConnection.class.isAssignableFrom(connectionType)
 				|| connectionType.equals(StatefulConnection.class)) {
-			return connectionType.cast(client.connect(codec));
+
+			StatefulRedisClusterConnection<?, ?> connection = client.connect(codec);
+			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/DefaultLettuceClientConfiguration.java

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.ClientOptions;
+import io.lettuce.core.ReadFrom;
 import io.lettuce.core.resource.ClientResources;
 
 import java.time.Duration;
@@ -38,11 +39,13 @@ class DefaultLettuceClientConfiguration implements LettuceClientConfiguration {
 	private final Optional<ClientResources> clientResources;
 	private final Optional<ClientOptions> clientOptions;
 	private final Optional<String> clientName;
+	private final Optional<ReadFrom> readFrom;
 	private final Duration timeout;
 	private final Duration shutdownTimeout;
 
 	DefaultLettuceClientConfiguration(boolean useSsl, boolean verifyPeer, boolean startTls,
 			@Nullable ClientResources clientResources, @Nullable ClientOptions clientOptions, @Nullable String clientName,
+			@Nullable ReadFrom readFrom,
 			Duration timeout, Duration shutdownTimeout) {
 
 		this.useSsl = useSsl;
@@ -51,6 +54,7 @@ class DefaultLettuceClientConfiguration implements LettuceClientConfiguration {
 		this.clientResources = Optional.ofNullable(clientResources);
 		this.clientOptions = Optional.ofNullable(clientOptions);
 		this.clientName = Optional.ofNullable(clientName);
+		this.readFrom = Optional.ofNullable(readFrom);
 		this.timeout = timeout;
 		this.shutdownTimeout = shutdownTimeout;
 	}
@@ -109,6 +113,15 @@ public Optional<String> getClientName() {
 		return clientName;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getReadFrom()
+	 */
+	@Override
+	public Optional<ReadFrom> getReadFrom() {
+		return readFrom;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getTimeout()

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

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.ClientOptions;
+import io.lettuce.core.ReadFrom;
 import io.lettuce.core.resource.ClientResources;
 
 import java.time.Duration;
@@ -96,6 +97,15 @@ public Optional<String> getClientName() {
 		return clientConfiguration.getClientName();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getReadFrom()
+	 */
+	@Override
+	public Optional<ReadFrom> getReadFrom() {
+		return clientConfiguration.getReadFrom();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getCommandTimeout()

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

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.ClientOptions;
+import io.lettuce.core.ReadFrom;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.resource.ClientResources;
 
@@ -38,6 +39,7 @@
  * <li>Optional {@link ClientResources}</li>
  * <li>Optional {@link ClientOptions}</li>
  * <li>Optional client name</li>
+ * <li>Optional {@link ReadFrom}. Enables Master/Slave operations if configured.</li>
  * <li>Client {@link Duration timeout}</li>
  * <li>Shutdown {@link Duration timeout}</li>
  * </ul>
@@ -82,6 +84,12 @@ public interface LettuceClientConfiguration {
 	 */
 	Optional<String> getClientName();
 
+	/**
+	 * @return the optional {@link io.lettuce.core.ReadFrom} setting.
+	 * @since 2.1
+	 */
+	Optional<ReadFrom> getReadFrom();
+
 	/**
 	 * @return the timeout.
 	 */
@@ -118,6 +126,8 @@ static LettuceClientConfigurationBuilder builder() {
 	 * <dd>none</dd>
 	 * <dt>Client name</dt>
 	 * <dd>none</dd>
+	 * <dt>Read From</dt>
+	 * <dd>none</dd>
 	 * <dt>Connect Timeout</dt>
 	 * <dd>60 Seconds</dd>
 	 * <dt>Shutdown Timeout</dt>
@@ -142,6 +152,7 @@ class LettuceClientConfigurationBuilder {
 		@Nullable ClientResources clientResources;
 		@Nullable ClientOptions clientOptions;
 		@Nullable String clientName;
+		@Nullable ReadFrom readFrom;
 		Duration timeout = Duration.ofSeconds(RedisURI.DEFAULT_TIMEOUT);
 		Duration shutdownTimeout = Duration.ofMillis(100);
 
@@ -188,6 +199,22 @@ public LettuceClientConfigurationBuilder clientOptions(ClientOptions clientOptio
 			return this;
 		}
 
+		/**
+		 * Configure {@link ReadFrom}. Enables Master/Slave operations if configured.
+		 *
+		 * @param readFrom must not be {@literal null}.
+		 * @return {@literal this} builder.
+		 * @throws IllegalArgumentException if clientOptions is {@literal null}.
+		 * @since 2.1
+		 */
+		public LettuceClientConfigurationBuilder readFrom(ReadFrom readFrom) {
+
+			Assert.notNull(readFrom, "ReadFrom must not be null!");
+
+			this.readFrom = readFrom;
+			return this;
+		}
+
 		/**
 		 * Configure a {@code clientName} to be set with {@code CLIENT SETNAME}.
 		 *
@@ -242,7 +269,7 @@ public LettuceClientConfigurationBuilder shutdownTimeout(Duration shutdownTimeou
 		public LettuceClientConfiguration build() {
 
 			return new DefaultLettuceClientConfiguration(useSsl, verifyPeer, startTls, clientResources, clientOptions,
-					clientName, timeout, shutdownTimeout);
+					clientName, readFrom, timeout, shutdownTimeout);
 		}
 	}
 

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

@@ -17,6 +17,7 @@
 
 import io.lettuce.core.AbstractRedisClient;
 import io.lettuce.core.ClientOptions;
+import io.lettuce.core.ReadFrom;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisException;
 import io.lettuce.core.RedisURI;
@@ -887,7 +888,7 @@ protected StatefulConnection<ByteBuffer, ByteBuffer> getSharedReactiveConnection
 
 	private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
 
-		LettuceConnectionProvider connectionProvider = doConnectionProvider(client, codec);
+		LettuceConnectionProvider connectionProvider = doCreateConnectionProvider(client, codec);
 
 		if (this.clientConfiguration instanceof LettucePoolingClientConfiguration) {
 			return new LettucePoolingConnectionProvider(connectionProvider,
@@ -897,13 +898,27 @@ private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient c
 		return connectionProvider;
 	}
 
-	private LettuceConnectionProvider doConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
+	/**
+	 * Create a {@link LettuceConnectionProvider} given {@link AbstractRedisClient} and {@link RedisCodec}. Configuration
+	 * of this connection factory specifies the type of the created connection provider. This method creates either a
+	 * {@link LettuceConnectionProvider} for either {@link RedisClient} or {@link RedisClusterClient}. Subclasses may
+	 * override this method to decorate the connection provider.
+	 *
+	 * @param client either {@link RedisClient} or {@link RedisClusterClient}, must not be {@literal null}.
+	 * @param codec used for connection creation, must not be {@literal null}. By default, a {@code byte[]} codec.
+	 *          Reactive connections require a {@link java.nio.ByteBuffer} codec.
+	 * @return the connection provider.
+	 * @since 2.1
+	 */
+	protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
+
+		ReadFrom readFrom = getClientConfiguration().getReadFrom().orElse(null);
 
 		if (isClusterAware()) {
-			return new ClusterConnectionProvider((RedisClusterClient) client, codec);
+			return new ClusterConnectionProvider((RedisClusterClient) client, codec, readFrom);
 		}
 
-		return new StandaloneConnectionProvider((RedisClient) client, codec);
+		return new StandaloneConnectionProvider((RedisClient) client, codec, readFrom);
 	}
 
 	private AbstractRedisClient createClient() {
@@ -1202,6 +1217,15 @@ public Optional<ClientOptions> getClientOptions() {
 			return Optional.empty();
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getReadFrom()
+		 */
+		@Override
+		public Optional<ReadFrom> getReadFrom() {
+			return Optional.empty();
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getClientName()