Consider intermediate command interface in ConnectionSplittingInterceptor.

We now consider requests to command API objects such as RedisConnection.keyCommands() in ConnectionSplittingInterceptor to identify the correct command and route it accordingly.

Closes: #2886
Original Pull Request: #2887

Author: mp911de

src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java

@@ -32,6 +32,8 @@
 import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Helper class that provides static methods for obtaining {@link RedisConnection} from a
@@ -320,8 +322,8 @@ private static RedisConnection getTargetConnection(RedisConnection connection) {
 	 */
 	public static void unbindConnection(RedisConnectionFactory factory) {
 
-		RedisConnectionHolder connectionHolder =
-				(RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);
+		RedisConnectionHolder connectionHolder = (RedisConnectionHolder) TransactionSynchronizationManager
+				.getResource(factory);
 
 		if (connectionHolder == null) {
 			return;
@@ -358,7 +360,8 @@ public static void unbindConnection(RedisConnectionFactory factory) {
 	 * @param connectionFactory Redis connection factory that the connection was created with
 	 * @return whether the connection is transactional or not
 	 */
-	public static boolean isConnectionTransactional(RedisConnection connection, RedisConnectionFactory connectionFactory) {
+	public static boolean isConnectionTransactional(RedisConnection connection,
+			RedisConnectionFactory connectionFactory) {
 
 		Assert.notNull(connectionFactory, "No RedisConnectionFactory specified");
 
@@ -448,9 +451,16 @@ public void afterCompletion(int status) {
 	static class ConnectionSplittingInterceptor implements MethodInterceptor {
 
 		private final RedisConnectionFactory factory;
+		private final @Nullable String commandInterface;
 
 		public ConnectionSplittingInterceptor(RedisConnectionFactory factory) {
 			this.factory = factory;
+			this.commandInterface = null;
+		}
+
+		public ConnectionSplittingInterceptor(RedisConnectionFactory factory, String commandInterface) {
+			this.factory = factory;
+			this.commandInterface = commandInterface;
 		}
 
 		@Override
@@ -465,6 +475,22 @@ public Object intercept(Object obj, Method method, Object[] args) throws Throwab
 				return obj;
 			}
 
+			Class<?> returnType = method.getReturnType();
+			String returnTypeName = returnType.getSimpleName();
+
+			// bridge keyCommands etc. to defer target invocations
+			if (returnType.isInterface() && returnType.getPackageName().equals("org.springframework.data.redis.connection")
+					&& returnTypeName.startsWith("Redis") && returnTypeName.endsWith("Commands")) {
+
+				ProxyFactory proxyFactory = new ProxyFactory(ReflectionUtils.invokeMethod(method, obj));
+
+				proxyFactory.addAdvice(new ConnectionSplittingInterceptor(factory, method.getName()));
+				proxyFactory.addInterface(RedisConnectionProxy.class);
+				proxyFactory.addInterface(returnType);
+
+				return proxyFactory.getProxy();
+			}
+
 			RedisCommand commandToExecute = RedisCommand.failsafeCommandLookup(method.getName());
 
 			if (isPotentiallyThreadBoundCommand(commandToExecute)) {
@@ -481,9 +507,15 @@ public Object intercept(Object obj, Method method, Object[] args) throws Throwab
 			}
 
 			RedisConnection connection = factory.getConnection();
-
+			Object target = connection;
 			try {
-				return invoke(method, connection, args);
+
+				if (StringUtils.hasText(commandInterface)) {
+					target = ReflectionUtils.invokeMethod(ReflectionUtils.findMethod(RedisConnection.class, commandInterface),
+							connection);
+				}
+
+				return invoke(method, target, args);
 			} finally {
 				// properly close the unbound connection after executing command
 				if (!connection.isClosed()) {

src/test/java/org/springframework/data/redis/connection/AbstractTransactionalTestBase.java

@@ -32,6 +32,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 import org.springframework.test.annotation.Rollback;
@@ -132,6 +133,18 @@ public void valueOperationSetShouldBeCommittedCorrectly() {
 		}
 	}
 
+	@Rollback(false)
+	@Test // GH-2886
+	public void shouldReturnReadOnlyCommandResultInTransaction() {
+
+		RedisTemplate<String, String> template = new RedisTemplate<>();
+		template.setConnectionFactory(factory);
+		template.setEnableTransactionSupport(true);
+		template.afterPropertiesSet();
+
+		assertThat(template.hasKey("foo")).isFalse();
+	}
+
 	@Test // DATAREDIS-548
 	@Transactional(readOnly = true)
 	public void valueOperationShouldWorkWithReadOnlyTransactions() {

src/test/java/org/springframework/data/redis/core/RedisConnectionUtilsUnitTests.java

@@ -24,6 +24,7 @@
 
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisKeyCommands;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
@@ -201,6 +202,45 @@ void bindConnectionShouldNotBindConnectionToTransaction() {
 		assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse();
 	}
 
+	@Test // GH-2886
+	void connectionProxyShouldInvokeReadOnlyMethods() {
+
+		TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager());
+
+		byte[] anyBytes = new byte[] { 1, 2, 3 };
+		when(connectionMock2.exists(anyBytes)).thenReturn(true);
+
+		template.executeWithoutResult(status -> {
+
+			RedisConnection connection = RedisConnectionUtils.getConnection(factoryMock, true);
+
+			assertThat(connection.exists(anyBytes)).isEqualTo(true);
+		});
+	}
+
+	@Test // GH-2886
+	void connectionProxyShouldConsiderCommandInterfaces() {
+
+		TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager());
+
+		byte[] anyBytes = new byte[] { 1, 2, 3 };
+
+		RedisKeyCommands commandsMock = mock(RedisKeyCommands.class);
+
+		when(connectionMock1.keyCommands()).thenReturn(commandsMock);
+		when(connectionMock2.keyCommands()).thenReturn(commandsMock);
+		when(commandsMock.exists(anyBytes)).thenReturn(true);
+		when(commandsMock.del(anyBytes)).thenReturn(42L);
+
+		template.executeWithoutResult(status -> {
+
+			RedisConnection connection = RedisConnectionUtils.getConnection(factoryMock, true);
+
+			assertThat(connection.keyCommands().exists(anyBytes)).isEqualTo(true);
+			assertThat(connection.keyCommands().del(anyBytes)).isEqualTo(42L);
+		});
+	}
+
 	static class DummyTransactionManager extends AbstractPlatformTransactionManager {
 
 		@Override