Fix Lettuce compareAndSet

DATAREDIS-122

- exec() now returns null if value modified by other thread

- Use same connection for watch/unwatch and txs

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

@@ -489,10 +489,10 @@ public class LettuceConnection implements RedisConnection {
 		isMulti = false;
 		try {
 			if (isPipelined()) {
-				pipeline(asyncTxConn.discard());
+				pipeline(getAsyncTxConnection().discard());
 				return;
 			}
-			txConn.discard();
+			getTxConnection().discard();
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -503,10 +503,14 @@ public class LettuceConnection implements RedisConnection {
 		isMulti = false;
 		try {
 			if (isPipelined()) {
-				 asyncTxConn.exec();
+				 getAsyncTxConnection().exec();
 				 return null;
 			}
-			return txConn.exec();
+			List<Object> results = getTxConnection().exec();
+			if(results.isEmpty()) {
+				return null;
+			}
+			return results;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -572,17 +576,11 @@ public class LettuceConnection implements RedisConnection {
 		}
 		isMulti = true;
 		try {
-			if(asyncTxConn == null) {
-				asyncTxConn = client.connectAsync(LettuceUtils.CODEC);
-			}
-			if(txConn == null) {
-				txConn = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(asyncTxConn);
-			}
 			if (isPipelined()) {
-				pipeline(asyncTxConn.multi());
+				pipeline(getAsyncTxConnection().multi());
 				return;
 			}
-			txConn.multi();
+			getTxConnection().multi();
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -688,10 +686,10 @@ public class LettuceConnection implements RedisConnection {
 	public void unwatch() {
 		try {
 			if (isPipelined()) {
-				pipeline(getAsyncConnection().unwatch());
+				pipeline(getAsyncTxConnection().unwatch());
 				return;
 			}
-			getConnection().unwatch();
+			getTxConnection().unwatch();
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -701,11 +699,11 @@ public class LettuceConnection implements RedisConnection {
 	public void watch(byte[]... keys) {
 		try {
 			if (isPipelined()) {
-				pipeline(getAsyncConnection().watch(keys));
+				pipeline(getAsyncTxConnection().watch(keys));
 				return;
 			}
 			else {
-				getConnection().watch(keys);
+				getTxConnection().watch(keys);
 			}
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
@@ -1922,21 +1920,21 @@ public class LettuceConnection implements RedisConnection {
 
 	private RedisAsyncConnection<byte[], byte[]> getAsyncConnection() {
 		if(isQueueing()) {
-			return asyncTxConn;
+			return getAsyncTxConnection();
 		}
 		return asyncConn;
 	}
 
 	private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getConnection() {
 		if(isQueueing()) {
-			return txConn;
+			return getTxConnection();
 		}
 		return con;
 	}
 
 	private RedisAsyncConnection<byte[], byte[]> getAsyncBlockingConnection() {
 		if(isQueueing()) {
-			return asyncTxConn;
+			return getAsyncTxConnection();
 		}
 		if(asyncBlockingConn ==  null) {
 			asyncBlockingConn = client.connectAsync(LettuceUtils.CODEC);
@@ -1946,11 +1944,25 @@ public class LettuceConnection implements RedisConnection {
 
 	private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getBlockingConnection() {
 		if(isQueueing()) {
-			return txConn;
+			return getTxConnection();
 		}
 		if(blockingConn ==  null) {
 			blockingConn = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(getAsyncBlockingConnection());
 		}
 		return blockingConn;
 	}
+
+	private RedisAsyncConnection<byte[], byte[]> getAsyncTxConnection() {
+		if(asyncTxConn == null) {
+			asyncTxConn = client.connectAsync(LettuceUtils.CODEC);
+		}
+		return asyncTxConn;
+	}
+
+	private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getTxConnection() {
+		if(txConn == null) {
+			txConn = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(getAsyncTxConnection());
+		}
+		return txConn;
+	}
 }

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

@@ -400,7 +400,6 @@ public abstract class AbstractConnectionIntegrationTests {
 	public void testWatch() throws Exception {
 		connection.set("testitnow", "willdo");
 		connection.watch("testitnow".getBytes());
-		connection.get("testitnow");
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
 				connectionFactory.getConnection());
 		conn2.set("testitnow", "something");

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

@@ -16,6 +16,9 @@
 
 package org.springframework.data.redis.connection.lettuce;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
 import java.util.List;
 
 import org.junit.Ignore;
@@ -26,8 +29,6 @@ import org.springframework.data.redis.connection.AbstractConnectionIntegrationTe
 import org.springframework.data.redis.connection.DefaultStringRedisConnection;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
 
 /**
  * Integration test of {@link LettuceConnection}
@@ -40,14 +41,6 @@ import static org.junit.Assert.fail;
 @ContextConfiguration
 public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegrationTests {
 
-	@Ignore("DATAREDIS-122 exec never returns null")
-	public void testWatch() throws Exception {
-	}
-
-	@Ignore("DATAREDIS-122 exec never returns null")
-	public void testUnwatch() throws Exception {
-	}
-
 	@Test
 	public void testMultiThreadsOneBlocking() throws Exception {
 		Thread th = new Thread(new Runnable() {
@@ -88,6 +81,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra
 		assertEquals("delay", conn2.get("txs1"));
 	}
 
+	@Ignore("DATAREDIS-189 Lettuce exec without multi times out waiting for a reply instead of throwing Exception")
 	@Test
 	public void testCloseInTransaction() {
 		connection.multi();

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

@@ -59,17 +59,44 @@ public class LettuceConnectionPipelineIntegrationTests extends
 	public void testMultiDiscard() {
 	}
 
-	@Ignore("DATAREDIS-122 exec never returns null")
-	public void testWatch() throws Exception {
-	}
-
-	@Ignore("DATAREDIS-122 exec never returns null")
-	public void testUnwatch() throws Exception {
-	}
-
 	// Overrides, usually due to return values being Long vs Boolean or Set vs
 	// List
 
+	@Test
+	public void testWatch() throws Exception {
+		connection.set("testitnow", "willdo");
+		connection.watch("testitnow".getBytes());
+		//Give some time for watch to be asynch executed
+		Thread.sleep(500);
+		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
+				connectionFactory.getConnection());
+		conn2.set("testitnow", "something");
+		conn2.close();
+		connection.multi();
+		connection.set("testitnow", "somethingelse");
+		actual.add(connection.exec());
+		actual.add(connection.get("testitnow"));
+		verifyResults(Arrays.asList(new Object[] { null, "something" }), actual);
+	}
+
+	@Test
+	public void testUnwatch() throws Exception {
+		connection.set("testitnow", "willdo");
+		connection.watch("testitnow".getBytes());
+		connection.unwatch();
+		connection.multi();
+		//Give some time for unwatch to be asynch executed
+		Thread.sleep(500);
+		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
+				connectionFactory.getConnection());
+		conn2.set("testitnow", "something");
+		connection.set("testitnow", "somethingelse");
+		connection.get("testitnow");
+		connection.exec();
+		List<Object> convertedResults = convertResults();
+		assertEquals(Arrays.asList(new Object[] {"somethingelse"}), convertedResults);
+	}
+
 	@Test
 	public void testBLPop() {
 		// Use a different synchronous connection to add items to the list, else blpop may

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

@@ -15,7 +15,8 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
-import java.util.Arrays;
+import static org.junit.Assert.assertNull;
+
 import java.util.List;
 
 import org.junit.Ignore;
@@ -87,10 +88,9 @@ public class LettuceConnectionTransactionIntegrationTests extends
 
 	@Test
 	public void exceptionExecuteNative() throws Exception {
-		actual.add(connection.execute("ZadD", getClass() + "#foo\t0.90\titem"));
-		// Syntax error on queued commands are swallowed and no results are
-		// returned
-		verifyResults(Arrays.asList(new Object[] {}), actual);
+		connection.execute("ZadD", getClass() + "#foo\t0.90\titem");
+		// Syntax error on queued commands are swallowed and no results are returned
+		assertNull(getResults());
 	}
 
 	protected void initConnection() {

src/test/java/org/springframework/data/redis/connection/rjc/RjcConnectionPipelineIntegrationTests.java

@@ -170,6 +170,8 @@ public class RjcConnectionPipelineIntegrationTests extends
 	public void testWatch() throws Exception {
 		connection.set("testitnow", "willdo");
 		connection.watch("testitnow".getBytes());
+		//Give some time for watch to be asynch executed
+		Thread.sleep(500);
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
 				connectionFactory.getConnection());
 		conn2.set("testitnow", "something");
@@ -188,6 +190,8 @@ public class RjcConnectionPipelineIntegrationTests extends
 		connection.set("testitnow", "willdo");
 		connection.watch("testitnow".getBytes());
 		connection.unwatch();
+		//Give some time for unwatch to be asynch executed
+		Thread.sleep(500);
 		connection.multi();
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
 				connectionFactory.getConnection());

src/test/java/org/springframework/data/redis/support/BoundKeyOperationsTest.java

@@ -31,7 +31,6 @@ import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
 import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.connection.ConnectionUtils;
-import org.springframework.data.redis.connection.jredis.JredisConnectionFactory;
 import org.springframework.data.redis.core.BoundKeyOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.support.collections.ObjectFactory;
@@ -70,6 +69,8 @@ public class BoundKeyOperationsTest {
 
 	@Test
 	public void testRename() throws Exception {
+		// DATAREDIS-188 Infinite loop renaming a non-existent Collection when using Lettuce
+		assumeTrue(!ConnectionUtils.isLettuce(template.getConnectionFactory()));
 		Object key = keyOps.getKey();
 		assertNotNull(key);
 		// RedisAtomicInteger/Long need to be reset, as they may be created

src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicTests.java

@@ -151,9 +151,8 @@ public class RedisAtomicTests {
 	// DATAREDIS-108
 	@Test
 	public void testCompareSet() throws Exception {
-		// Txs not supported in Jredis, Lettuce checkAndSet not working DATAREDIS-122,
-		// SRP checkAndSet not working DATAREDIS-123
-		assumeTrue(!ConnectionUtils.isJredis(factory) && !ConnectionUtils.isLettuce(factory) && !ConnectionUtils.isSrp(factory));
+		// Txs not supported in Jredis, SRP checkAndSet not working DATAREDIS-123
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !ConnectionUtils.isSrp(factory));
 		final AtomicBoolean alreadySet = new AtomicBoolean(false);
 		final int NUM = 50;
 		final String KEY = getClass().getSimpleName() + ":atomic:counter";