Lettuce closePipeline should return List of tx results

and strip individual op results

DATAREDIS-223

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

@@ -683,11 +683,10 @@ public abstract class AbstractConnectionIntegrationTests {
 	public void testMultiExec() throws Exception {
 		connection.multi();
 		connection.set("key", "value");
-		actual.add(connection.get("key"));
+		connection.get("key");
 		actual.add(connection.exec());
 		List<Object> results = getResults();
-		assertNull(results.get(0));
-		List<Object> execResults = (List<Object>) results.get(1);
+		List<Object> execResults = (List<Object>) results.get(0);
 		assertEquals(2, execResults.size());
 		assertEquals("value", new String((byte[]) execResults.get(1)));
 		assertEquals("value", connection.get("key"));
@@ -700,6 +699,22 @@ public abstract class AbstractConnectionIntegrationTests {
 		testMultiExec();
 	}
 
+	@Test(expected=RedisSystemException.class)
+	public void testExecWithoutMulti() {
+		connection.exec();
+		getResults();
+	}
+
+	@Test(expected=RedisSystemException.class)
+	public void testErrorInTx() {
+		connection.multi();
+		connection.set("foo","bar");
+		// Try to do a list op on a value
+		connection.lPop("foo");
+		connection.exec();
+		getResults();
+	}
+
 	@Test
 	public void testMultiDiscard() throws Exception {
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
@@ -720,6 +735,8 @@ public abstract class AbstractConnectionIntegrationTests {
 	public void testWatch() throws Exception {
 		connection.set("testitnow", "willdo");
 		connection.watch("testitnow".getBytes());
+		//Give some time for watch to be asynch executed in extending tests
+		Thread.sleep(500);
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
 				connectionFactory.getConnection());
 		conn2.set("testitnow", "something");
@@ -738,15 +755,16 @@ public abstract class AbstractConnectionIntegrationTests {
 		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");
-		actual.add(connection.get("testitnow"));
+		connection.get("testitnow");
 		actual.add(connection.exec());
 		List<Object> results = getResults();
-		assertNull(results.get(0));
-		List<Object> execResults = (List<Object>) results.get(1);
+		List<Object> execResults = (List<Object>) results.get(0);
 		assertEquals("somethingelse", new String((byte[]) execResults.get(1)));
 	}
 

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

@@ -17,7 +17,6 @@
 package org.springframework.data.redis.connection;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
@@ -66,6 +65,16 @@ abstract public class AbstractConnectionPipelineIntegrationTests extends
 	public void testPubSubWithPatterns() throws Exception {
 	}
 
+	@Test(expected=RedisPipelineException.class)
+	public void testExecWithoutMulti() {
+		super.testExecWithoutMulti();
+	}
+
+	@Test(expected=RedisPipelineException.class)
+	public void testErrorInTx() {
+		super.testErrorInTx();
+	}
+
 	@Test(expected = RedisPipelineException.class)
 	public void exceptionExecuteNative() throws Exception {
 		super.exceptionExecuteNative();
@@ -115,36 +124,6 @@ abstract public class AbstractConnectionPipelineIntegrationTests extends
 		assertTrue(results.isEmpty());
 	}
 
-	@Test
-	public void testMultiExec() throws Exception {
-		connection.multi();
-		connection.set("key", "value");
-		assertNull(connection.get("key"));
-		assertNull(connection.exec());
-		// We expect pipelined exec results to be flattened into results of closePipeline
-		List<Object> results = getResults();
-		assertEquals(1,results.size());
-		assertEquals("value",new String((byte[])results.get(0)));
-	}
-
-	@Test
-	public void testUnwatch() throws Exception {
-		connection.set("testitnow", "willdo");
-		connection.watch("testitnow".getBytes());
-		connection.unwatch();
-		connection.multi();
-		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
-				connectionFactory.getConnection());
-		conn2.set("testitnow", "something");
-		connection.set("testitnow", "somethingelse");
-		connection.get("testitnow");
-		connection.exec();
-		// We expect pipelined exec results to be flattened into results of closePipeline
-		List<Object> results = getResults();
-		assertEquals(1,results.size());
-		assertEquals("somethingelse",new String((byte[])results.get(0)));
-	}
-
 	protected void initConnection() {
 		connection.openPipeline();
 	}

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

@@ -44,6 +44,16 @@ abstract public class AbstractConnectionTransactionIntegrationTests extends
 	public void testWatch() {
 	}
 
+	@Ignore
+	@Test
+	public void testExecWithoutMulti() {
+	}
+
+	@Ignore
+	@Test
+	public void testErrorInTx() {
+	}
+
 	/*
 	 * Using blocking ops inside a tx does not make a lot of sense as it would require blocking the
 	 * entire server in order to execute the block atomically, which in turn does not allow other
@@ -399,8 +409,19 @@ abstract public class AbstractConnectionTransactionIntegrationTests extends
 		if (actual == null) {
 			return null;
 		}
+		return convertResults(actual);
+	}
+
+	protected List<Object> getResultsNoConversion() {
+		return connection.exec();
+	}
+
+	protected List<Object> convertResults(List<Object> results) {
+		if(results == null) {
+			return null;
+		}
 		List<Object> serializedResults = new ArrayList<Object>();
-		for (Object result : actual) {
+		for (Object result : results) {
 			Object convertedResult = convertResult(result);
 			if (convertedResult instanceof Exception) {
 				throw convertException((Exception) convertedResult);

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

@@ -149,6 +149,18 @@ public class DefaultStringRedisConnectionPipelineTests extends DefaultStringRedi
 		super.testExec();
 	}
 
+	@Test
+	public void testTxResultsNotPipelined() {
+		doReturn(true).when(nativeConnection).isQueueing();
+		List<Object> results = Arrays.asList(new Object[] { bar, 8l });
+		doReturn(Arrays.asList(new Object[] { results })).when(nativeConnection).closePipeline();
+		doReturn(8l).when(nativeConnection).lLen(fooBytes);
+		connection.lLen(fooBytes);
+		connection.exec();
+		// closePipeline should only return the results of exec, not of llen
+		verifyResults(Arrays.asList(new Object[] {results}));
+	}
+
 	@Test
 	public void testExistsBytes() {
 		doReturn(Arrays.asList(new Object[] { true })).when(nativeConnection).closePipeline();

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java

@@ -16,6 +16,10 @@
 
 package org.springframework.data.redis.connection.jedis;
 
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+
 import org.junit.After;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -26,6 +30,8 @@ import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import redis.clients.jedis.exceptions.JedisDataException;
+
 /**
  * Integration test of {@link JedisConnection}
  *
@@ -51,6 +57,18 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 		connection = null;
 	}
 
+	@Test
+	public void testErrorInTx() {
+		connection.multi();
+		connection.set("foo","bar");
+		// Try to do a list op on a value
+		connection.lPop("foo");
+		List<Object> results = connection.exec();
+		// Jedis puts the Exception in the exec results instead of throwing an Exception on exec
+		// This should be fixed in DATAREDIS-208 when we convert results of tx.exec()
+		assertTrue(results.get(1) instanceof JedisDataException);
+	}
+
 	@Test(expected=UnsupportedOperationException.class)
 	public void testPExpire() {
 		super.testPExpire();
@@ -208,4 +226,9 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 	public void testEvalShaNotFound() {
 		connection.evalSha("somefakesha", ReturnType.VALUE, 2, "key1", "key2");
 	}
+
+	@Test(expected=InvalidDataAccessApiUsageException.class)
+	public void testExecWithoutMulti() {
+		super.testExecWithoutMulti();
+	}
 }

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java

@@ -49,7 +49,7 @@ public class JedisConnectionPipelineIntegrationTests extends
 		connection = null;
 	}
 
-	@Ignore("DATAREDIS-143 Pipeline tries to return String instead of List<String>")
+	@Ignore("Jedis issue: Pipeline tries to return String instead of List<String>")
 	public void testGetConfig() {
 	}
 
@@ -61,15 +61,15 @@ public class JedisConnectionPipelineIntegrationTests extends
 	public void testUnwatch() {
 	}
 
-	@Ignore("DATAREDIS-143 Pipeline tries to return List<String> instead of Long on sort")
+	@Ignore("https://github.com/xetorthio/jedis/pull/389 Pipeline tries to return List<String> instead of Long on sort")
 	public void testSortStore() {
 	}
 
-	@Ignore("DATAREDIS-143 Pipeline tries to return Long instead of List<String> on sort with no params")
+	@Ignore("Jedis issue: Pipeline tries to return Long instead of List<String> on sort with no params")
 	public void testSortNullParams() {
 	}
 
-	@Ignore("DATAREDIS-143 Pipeline tries to return Long instead of List<String> on sort with no params")
+	@Ignore("https://github.com/xetorthio/jedis/pull/389 Pipeline tries to return Long instead of List<String> on sort with no params")
 	public void testSortStoreNullParams() {
 	}
 
@@ -81,6 +81,11 @@ public class JedisConnectionPipelineIntegrationTests extends
 	public void testMultiDiscard() {
 	}
 
+	@Ignore("DATAREDIS-143 Jedis ClassCastExceptions closing pipeline on certain ops")
+	@Test
+	public void testErrorInTx() {
+	}
+
 	// Unsupported Ops
 	@Test(expected=UnsupportedOperationException.class)
 	public void testPExpire() {

src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java

@@ -160,6 +160,16 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat
 		super.testUnwatch();
 	}
 
+	@Test(expected=UnsupportedOperationException.class)
+	public void testErrorInTx() {
+		super.testErrorInTx();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testExecWithoutMulti() {
+		super.testExecWithoutMulti();
+	}
+
 	@Test(expected = UnsupportedOperationException.class)
 	public void testBLPop() {
 		super.testBLPop();

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

@@ -23,6 +23,7 @@ import static org.junit.Assume.assumeTrue;
 import static org.springframework.data.redis.SpinBarrier.waitFor;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
@@ -43,6 +44,7 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.lambdaworks.redis.RedisAsyncConnection;
+import com.lambdaworks.redis.RedisException;
 
 /**
  * Integration test of {@link LettuceConnection}
@@ -284,4 +286,16 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra
 			conn2.close();
 		}
 	}
+
+	@Test
+	public void testErrorInTx() {
+		connection.multi();
+		connection.set("foo","bar");
+		// Try to do a list op on a value
+		connection.lPop("foo");
+		List<Object> execResults = connection.exec();
+		// Lettuce puts the Exception in the exec results instead of throwing an Exception on exec
+		// This should be fixed in DATAREDIS-208 when we convert results of tx.exec()
+		assertTrue(execResults.get(1) instanceof RedisException);
+	}
 }

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

@@ -24,7 +24,6 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.dao.DataAccessException;
@@ -40,6 +39,8 @@ import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import com.lambdaworks.redis.RedisException;
+
 /**
  * Integration test of {@link LettuceConnection} pipeline functionality
  *
@@ -50,10 +51,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @ContextConfiguration("LettuceConnectionIntegrationTests-context.xml")
 public class LettuceConnectionPipelineIntegrationTests extends
 		AbstractConnectionPipelineIntegrationTests {
-
-	@Ignore("DATAREDIS-144 Lettuce closePipeline hangs with discarded transaction")
-	public void testMultiDiscard() {
-	}
 	
 	@Test(expected=UnsupportedOperationException.class)
 	public void testSelect() {
@@ -79,47 +76,6 @@ public class LettuceConnectionPipelineIntegrationTests extends
 		super.testSRandMemberCountNegative();
 	}
 
-	@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"));
-		// Exec results lost on flatten in closePipeline and also won't get converted
-		// by DefaultStringRedisConnection yet
-		List<Object> results = getResults();
-		assertEquals("something", new String((byte[])results.get(0)));
-	}
-
-	@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();
-		// Exec results lost on flatten in closePipeline and also won't get converted
-				// by DefaultStringRedisConnection yet
-		List<Object> results = getResults();
-		assertEquals(1,results.size());
-		assertEquals("somethingelse",new String((byte[])results.get(0)));;
-	}
-
 	@Test
 	@IfProfileValue(name = "runLongTests", value = "true")
 	public void testScriptKill() throws Exception{
@@ -173,4 +129,19 @@ public class LettuceConnectionPipelineIntegrationTests extends
 			conn2.close();
 		}
 	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testErrorInTx() {
+		connection.multi();
+		connection.set("foo","bar");
+		// Try to do a list op on a value
+		connection.lPop("foo");
+		connection.exec();
+		List<Object> results = getResults();
+		List<Object> execResults = (List<Object>)results.get(0);
+		// Lettuce puts the Exception in the exec results instead of throwing an Exception on exec
+		// This should be fixed in DATAREDIS-208 when we convert results of tx.exec()
+		assertTrue(execResults.get(1) instanceof RedisException);
+	}
 }

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

@@ -3,24 +3,7 @@ package org.springframework.data.redis.connection.lettuce;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.junit.Ignore;
-import org.junit.Test;
-import org.springframework.core.convert.converter.Converter;
-import org.springframework.data.redis.connection.DefaultStringTuple;
-import org.springframework.data.redis.connection.RedisPipelineException;
-import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
-import org.springframework.data.redis.connection.ReturnType;
-import org.springframework.data.redis.connection.StringRedisConnection.StringTuple;
-import org.springframework.data.redis.connection.convert.SetConverter;
-import org.springframework.data.redis.serializer.SerializationUtils;
-import org.springframework.test.annotation.IfProfileValue;
 
 /**
  * Integration test of {@link LettuceConnection} transactions within a pipeline
@@ -29,257 +12,32 @@ import org.springframework.test.annotation.IfProfileValue;
  *
  */
 public class LettuceConnectionPipelineTxIntegrationTests extends
-		LettuceConnectionPipelineIntegrationTests {
-
-	private boolean convert=true;
-
-	private boolean convertToStringTuple=true;
-
-	@Ignore
-	public void testMultiDiscard() {
-	}
-
-	@Ignore
-	public void testMultiExec() {
-	}
-
-	@Ignore
-	public void testUnwatch() {
-	}
-
-	@Ignore
-	public void testWatch() {
-	}
-
-	/*
-	 * Using blocking ops inside a tx does not make a lot of sense as it would
-	 * require blocking the entire server in order to execute the block
-	 * atomically, which in turn does not allow other clients to perform a push
-	 * operation. *
-	 */
-
-	@Ignore
-	public void testBLPop() {
-	}
-
-	@Ignore
-	public void testBRPop() {
-	}
-
-	@Ignore
-	public void testBRPopLPush() {
-	}
-
-	@Ignore
-	public void testBLPopTimeout() {
-	}
-
-	@Ignore
-	public void testBRPopTimeout() {
-	}
-
-	@Ignore
-	public void testBRPopLPushTimeout() {
-	}
-
-	@Ignore("Pub/Sub not supported with transactions")
-	public void testPubSubWithNamedChannels() throws Exception {
-	}
-
-	@Ignore("Pub/Sub not supported with transactions")
-	public void testPubSubWithPatterns() throws Exception {
-	}
-
-	@Ignore
-	public void testNullKey() throws Exception {
-	}
-
-	@Ignore
-	public void testNullValue() throws Exception {
-	}
-
-	@Ignore
-	public void testHashNullKey() throws Exception {
-	}
-
-	@Ignore
-	public void testHashNullValue() throws Exception {
-	}
-
-	// Lettuce closePipeline() is now returning converted data types for executed txs, but
-	// DefaultStringRedisConnection isn't converting byte[]s from exec() yet.
-	// We implement the conversion in convertResult(), but convertResult's a bit over-eager in certain scenarios, hence these overrides
-
-	@Test(expected = RedisPipelineException.class)
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testRestoreExistingKey() {
-		convert = false;
-		super.testRestoreExistingKey();
-	}
-
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testScriptLoadEvalSha() {
-		convert = false;
-		super.testScriptLoadEvalSha();
-	}
-
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testEvalShaArrayStrings() {
-		convert = false;
-		super.testEvalShaArrayStrings();
-	}
-
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testEvalReturnString() {
-		convert = false;
-		super.testEvalReturnString();
-	}
-
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testEvalReturnArrayStrings() {
-		convert = false;
-		super.testEvalReturnArrayStrings();
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testEvalReturnArrayOKs() {
-		actual.add(connection.eval(
-				"return { redis.call('set','abc','ghk'),  redis.call('set','abc','lfdf')}",
-				ReturnType.MULTI, 0));
-		List<String> result = (List<String>) getResults().get(0);
-		assertEquals(Arrays.asList(new Object[] { "OK", "OK" }), result);
-	}
-
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testDumpAndRestore() {
-		convert = false;
-		connection.set("testing", "12");
-		actual.add(connection.dump("testing".getBytes()));
-		List<Object> results = getResults();
-		initConnection();
-		actual.add(connection.del("testing"));
-		actual.add((connection.get("testing")));
-		connection.restore("testing".getBytes(), 0, (byte[]) results.get(results.size() - 1));
-		actual.add(connection.get("testing"));
-		results = getResults();
-		assertEquals(3,results.size());
-		assertEquals(1l, results.get(0));
-		assertNull(results.get(1));
-		assertEquals("12", new String((byte[])results.get(2)));
-	}
-
-	@Test
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testRestoreTtl() {
-		convert = false;
-		super.testRestoreTtl();
-	}
-
-	@Test
-	public void testExecute() {
-		convert = false;
-		super.testExecute();
-	}
-
-	@Test
-	public void testExecuteNoArgs() {
-		convert = false;
-		super.testExecuteNoArgs();
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test
-	public void testZRevRangeByScoreOffsetCount() {
-		actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes()));
-		actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes()));
-		actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d, 0, 5));
-		assertEquals(new LinkedHashSet<String>(Arrays.asList(new String[] { "Bob", "James" })),(Set<String>) getResults().get(2));
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test
-	public void testZRevRangeByScore() {
-		actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes()));
-		actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes()));
-		actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d));
-		assertEquals(new LinkedHashSet<String>(Arrays.asList(new String[] { "Bob", "James" })), (Set<String>) getResults().get(2));
-	}
-
-	@Test
-	public void testZRevRangeByScoreWithScoresOffsetCount() {
-		convertToStringTuple = false;
-		super.testZRevRangeByScoreWithScoresOffsetCount();
-	}
-
-	@Test
-	public void testZRevRangeByScoreWithScores() {
-		convertToStringTuple = false;
-		super.testZRevRangeByScoreWithScores();
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	@IfProfileValue(name = "redisVersion", value = "2.6")
-	public void testScriptKill() {
-		// Impossible to call script kill in a tx because you can't issue the
-		// exec command while Redis is running a script
-		connection.scriptKill();
-	}
+		LettuceConnectionTransactionIntegrationTests {
 
 	protected void initConnection() {
 		connection.openPipeline();
 		connection.multi();
 	}
 
+	@SuppressWarnings("unchecked")
 	protected List<Object> getResults() {
 		assertNull(connection.exec());
 		List<Object> pipelined = connection.closePipeline();
-		List<Object> convertedResults = new ArrayList<Object>();
-		for(Object result: pipelined) {
-			convertedResults.add(convertResult(result));
-		}
-		return convertedResults;
+		// We expect only the results of exec to be in the closed pipeline
+		assertEquals(1,pipelined.size());
+		List<Object> txResults = (List<Object>)pipelined.get(0);
+		// Return exec results and this test should behave exactly like its superclass
+		return convertResults(txResults);
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	protected Object convertResult(Object result) {
-		if(!convert) {
-			return result;
-		}
-		if (result instanceof List && !(((List) result).isEmpty())
-				&& ((List) result).get(0) instanceof byte[]) {
-			return (SerializationUtils.deserialize((List<byte[]>) result, stringSerializer));
-		} else if (result instanceof byte[]) {
-			return (stringSerializer.deserialize((byte[]) result));
-		} else if (result instanceof Map
-				&& ((Map) result).keySet().iterator().next() instanceof byte[]) {
-			return (SerializationUtils.deserialize((Map) result, stringSerializer));
-		} else if (result instanceof Set && !(((Set) result).isEmpty())) {
-			Object firstResult = ((Set) result).iterator().next();
-			if(firstResult instanceof byte[]) {
-				return (SerializationUtils.deserialize((Set) result, stringSerializer));
-			}
-		}
-		if (result instanceof Set && !(((Set) result).isEmpty())
-				&& ((Set) result).iterator().next() instanceof Tuple) {
-			if(convertToStringTuple) {
-				return new SetConverter<Tuple, StringTuple>(new TupleConverter())
-					.convert((Set)result);
-			}
-		}
-		return result;
+	@SuppressWarnings("unchecked")
+	protected List<Object> getResultsNoConversion() {
+		assertNull(connection.exec());
+		List<Object> pipelined = connection.closePipeline();
+		// We expect only the results of exec to be in the closed pipeline
+		assertEquals(1,pipelined.size());
+		List<Object> txResults = (List<Object>)pipelined.get(0);
+		// Return exec results and this test should behave exactly like its superclass
+		return txResults;
 	}
-
-	private class TupleConverter implements Converter<Tuple, StringTuple> {
-		public StringTuple convert(Tuple source) {
-			return new DefaultStringTuple(source, stringSerializer.deserialize(source.getValue()));
-		}
-	}
-
 }

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

@@ -60,12 +60,9 @@ public class LettuceConnectionTransactionIntegrationTests extends
 
 	private boolean convertToStringTuple=true;
 
-	@Ignore
-	public void testMultiConnectionsOneInTx() {
-	}
-
-	@Ignore
-	public void testCloseBlockingOps() {
+	@Test
+	@Ignore("Exceptions on native execute are swallowed in tx")
+	public void exceptionExecuteNative() throws Exception {
 	}
 
 	// Native Lettuce returns ZSets as Lists
@@ -201,14 +198,6 @@ public class LettuceConnectionTransactionIntegrationTests extends
 		super.testBitOpNotMultipleSources();
 	}
 
-	@Test
-	public void exceptionExecuteNative() throws Exception {
-		connection.execute("ZadD", getClass() + "#foo\t0.90\titem");
-		// Syntax error on queued commands are swallowed and no results are
-		// returned
-		assertNull(getResults());
-	}
-
 	@Test
 	public void testGetRangeSetRange() {
 		connection.set("rangekey", "supercalifrag");
@@ -269,7 +258,7 @@ public class LettuceConnectionTransactionIntegrationTests extends
 	@IfProfileValue(name = "redisVersion", value = "2.6")
 	public void testEvalReturnSingleOK() {
 		actual.add(connection.eval("return redis.call('set','abc','ghk')", ReturnType.STATUS, 0));
-		assertEquals(Arrays.asList(new Object[] { "OK" }), connection.exec());
+		assertEquals(Arrays.asList(new Object[] { "OK" }), getResultsNoConversion());
 	}
 
 	@Test

src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineIntegrationTests.java

@@ -19,9 +19,11 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
 import java.util.Arrays;
+import java.util.List;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.AbstractConnectionPipelineIntegrationTests;
 import org.springframework.data.redis.connection.DefaultStringRedisConnection;
 import org.springframework.data.redis.connection.RedisPipelineException;
@@ -62,6 +64,38 @@ public class SrpConnectionPipelineIntegrationTests extends
 		testMultiExec();
 	}
 
+	@Test
+	public void testMultiExec() throws Exception {
+		connection.multi();
+		connection.set("key", "value");
+		connection.get("key");
+		actual.add(connection.exec());
+		List<Object> results = getResults();
+		// For the moment, SRP exec results are flattened into pipeline
+		assertEquals(1, results.size());
+		assertEquals("value", new String((byte[]) results.get(0)));
+		assertEquals("value", connection.get("key"));
+	}
+
+	@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");
+		actual.add(connection.exec());
+		List<Object> results = getResults();
+		// For the moment, SRP exec results are flattened into pipeline
+		assertEquals("somethingelse", new String((byte[]) results.get(0)));
+	}
+
 	// SRP sets results of all commands in the pipeline to RedisException if
 	// exec returns a
 	// null multi-bulk reply
@@ -81,6 +115,13 @@ public class SrpConnectionPipelineIntegrationTests extends
 		verifyResults(Arrays.asList(new Object[] { Arrays.asList(new Object[] { "OK", "OK" }) }));
 	}
 
+	@Test(expected=RedisSystemException.class)
+	public void testExecWithoutMulti() {
+		connection.exec();
+		// SRP throws an Exception right away on exec instead of once pipeline is closed
+		getResults();
+	}
+
 	@Test
 	public void testExecuteNoArgs() {
 		// SRP returns this as String while other drivers return as byte[]

src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionTransactionIntegrationTests.java

@@ -61,6 +61,16 @@ public class SrpConnectionTransactionIntegrationTests extends SrpConnectionInteg
 	public void testWatch() {
 	}
 
+	@Ignore
+	@Test
+	public void testExecWithoutMulti() {
+	}
+
+	@Ignore
+	@Test
+	public void testErrorInTx() {
+	}
+
 	/*
 	 * Using blocking ops inside a tx does not make a lot of sense as it would
 	 * require blocking the entire server in order to execute the block