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

We now allow configuration of ReadFrom settings using the Lettuce driver to enable slave reads. If ReadFrom is configured, we opt-in to Master/Slave connections instead of plain connections. Master/Slave connections route commands to the configured type of node depending on whether the command is a read or write command.

LettuceClientConfiguration configuration = LettuceClientConfiguration.builder().readFrom(ReadFrom.SLAVE).build();

ReadFrom is available for:

* Static Master/Slave Redis without Redis Sentinel
* Sentinel-Managed Master/Slave Redis
* Redis Cluster

ReadFrom is not configured for Pub/Sub connections or connections to the actual Sentinel servers.

Author: mp911de

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

@@ -15,11 +15,15 @@
  */
 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 lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
 
 /**
  * Connection provider for Cluster connections.
@@ -28,16 +32,12 @@
  * @author Christoph Strobl
  * @since 2.0
  */
+@RequiredArgsConstructor
 class ClusterConnectionProvider implements LettuceConnectionProvider {
 
 	private final RedisClusterClient client;
 	private final RedisCodec<?, ?> codec;
-
-	ClusterConnectionProvider(RedisClusterClient client, RedisCodec<?, ?> codec) {
-
-		this.client = client;
-		this.codec = codec;
-	}
+	private final Optional<ReadFrom> readFrom;
 
 	/*
 	 * (non-Javadoc)
@@ -52,7 +52,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;
 	}
@@ -91,8 +95,7 @@ public Optional<ClientResources> getClientResources() {
 		return clientResources;
 	}
 
-	/*
-	 * (non-Javadoc)
+	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getClientOptions()
 	 */
 	@Override
@@ -109,6 +112,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/LettuceClusterConnection.java

@@ -30,6 +30,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
@@ -74,7 +75,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	 * @param clusterClient must not be {@literal null}.
 	 */
 	public LettuceClusterConnection(RedisClusterClient clusterClient) {
-		this(new ClusterConnectionProvider(clusterClient, CODEC));
+		this(new ClusterConnectionProvider(clusterClient, CODEC, Optional.empty()));
 	}
 
 	/**
@@ -98,7 +99,7 @@ public LettuceClusterConnection(RedisClusterClient clusterClient, ClusterCommand
 	 * @since 2.0
 	 */
 	public LettuceClusterConnection(RedisClusterClient clusterClient, ClusterCommandExecutor executor, Duration timeout) {
-		this(new ClusterConnectionProvider(clusterClient, CODEC), executor, timeout);
+		this(new ClusterConnectionProvider(clusterClient, CODEC, Optional.empty()), executor, timeout);
 	}
 
 	/**

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

@@ -49,6 +49,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
@@ -224,7 +225,7 @@ public LettuceConnection(@Nullable StatefulRedisConnection<byte[], byte[]> share
 		if (pool != null) {
 			this.connectionProvider = new LettucePoolConnectionProvider(pool);
 		} else {
-			this.connectionProvider = new StandaloneConnectionProvider((RedisClient) client, CODEC);
+			this.connectionProvider = new StandaloneConnectionProvider((RedisClient) client, CODEC, Optional.empty());
 		}
 
 		this.asyncSharedConn = sharedConnection;

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;
@@ -822,7 +823,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,
@@ -832,12 +833,29 @@ 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
+	 * @see io.lettuce.core.codec.ByteArrayCodec
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection.ByteBufferCodec
+	 */
+	protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
+
+		Optional<ReadFrom> readFrom = getClientConfiguration().getReadFrom();
+
 		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() {
@@ -1124,6 +1142,14 @@ 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()

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

@@ -15,14 +15,20 @@
  */
 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 io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
 import lombok.RequiredArgsConstructor;
 
+import java.util.Optional;
+
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
 
 /**
@@ -35,6 +41,7 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 
 	private final RedisClient client;
 	private final RedisCodec<?, ?> codec;
+	private final Optional<ReadFrom> readFrom;
 
 	/*
 	 * (non-Javadoc)
@@ -53,7 +60,17 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 		}
 
 		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
-			return connectionType.cast(client.connect(codec));
+
+			return readFrom.map(it -> {
+
+				DirectFieldAccessor fieldAccessor = new DirectFieldAccessor(client);
+				RedisURI redisURI = RedisURI.class.cast(fieldAccessor.getPropertyValue("redisURI"));
+
+				StatefulRedisMasterSlaveConnection<?, ?> connection = MasterSlave.connect(client, codec, redisURI);
+				connection.setReadFrom(it);
+
+				return connectionType.cast(connection);
+			}).orElseGet(() -> connectionType.cast(client.connect(codec)));
 		}
 
 		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java

@@ -20,6 +20,7 @@
 import static org.hamcrest.core.IsNull.*;
 import static org.junit.Assert.*;
 
+import io.lettuce.core.ReadFrom;
 import io.lettuce.core.RedisException;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
@@ -362,6 +363,28 @@ public void factoryCreatesPooledConnections() {
 		factory.destroy();
 	}
 
+	@Test // DATAREDIS-580
+	public void factoryUsesMasterSlaveConnections() {
+
+		LettuceClientConfiguration configuration = LettuceTestClientConfiguration.builder().readFrom(ReadFrom.SLAVE)
+				.build();
+
+		LettuceConnectionFactory factory = new LettuceConnectionFactory(SettingsUtils.standaloneConfiguration(),
+				configuration);
+		factory.afterPropertiesSet();
+
+		RedisConnection connection = factory.getConnection();
+
+		try {
+			assertThat(connection.ping(), is(equalTo("PONG")));
+			assertThat(connection.info().getProperty("role"), is(equalTo("slave")));
+		} finally {
+			this.connection.close();
+		}
+
+		factory.destroy();
+	}
+
 	@Test // DATAREDIS-576
 	public void connectionAppliesClientName() {