DATAREDIS-1142 - Use Lettuce reactive cluster connection if configured.

Original Pull Request: #529

Author: andreacomo

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

@@ -90,6 +90,7 @@
  * @author Balázs Németh
  * @author Ruben Cervilla
  * @author Luis De Bello
+ * @author Andrea Como
  */
 public class LettuceConnectionFactory
 		implements InitializingBean, DisposableBean, RedisConnectionFactory, ReactiveRedisConnectionFactory {
@@ -425,6 +426,10 @@ protected LettuceClusterConnection doCreateLettuceClusterConnection(
 	@Override
 	public LettuceReactiveRedisConnection getReactiveConnection() {
 
+		if (isClusterAware()) {
+			return getReactiveClusterConnection();
+		}
+
 		return getShareNativeConnection()
 				? new LettuceReactiveRedisConnection(getSharedReactiveConnection(), reactiveConnectionProvider)
 				: new LettuceReactiveRedisConnection(reactiveConnectionProvider);

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

@@ -66,6 +66,7 @@
  * @author Balázs Németh
  * @author Ruben Cervilla
  * @author Luis De Bello
+ * @author Andrea Como
  */
 public class LettuceConnectionFactoryUnitTests {
 
@@ -939,6 +940,27 @@ public void maxRedirectsShouldBeSetOnClusterClientOptions() {
 		assertThat(options.getTimeoutOptions().isApplyConnectionTimeout()).isFalse();
 	}
 
+	@Test // DATAREDIS-1142
+	public void shouldFallbackToReactiveRedisClusterConnectionWhenGetReactiveConnectionWithClusterConfig() {
+
+		LettuceConnectionProvider connectionProviderMock = mock(LettuceConnectionProvider.class);
+		StatefulConnection<?, ?> statefulConnection = mock(StatefulConnection.class);
+		when(connectionProviderMock.getConnection(any())).thenReturn(statefulConnection);
+		LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig) {
+
+			@Override
+			protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClient client,
+					RedisCodec<?, ?> codec) {
+				return connectionProviderMock;
+			}
+		};
+		connectionFactory.afterPropertiesSet();
+
+		LettuceReactiveRedisConnection reactiveConnection = connectionFactory.getReactiveConnection();
+
+		assertThat(reactiveConnection).isInstanceOf(LettuceReactiveRedisClusterConnection.class);
+	}
+
 	@Data
 	@AllArgsConstructor
 	static class CustomRedisConfiguration implements RedisConfiguration, WithHostAndPort {