Fix Jedis ClassCastExceptions closing pipeline after tx

DATAREDIS-143

Use pipeline instance instead of transaction instance
to execute ops

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

@@ -106,6 +106,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		connection = null;
 	}
 	
+	@Test
 	public void testSelect() {
 		// Make sure this doesn't throw Exception
 		connection.select(1);
@@ -1838,6 +1839,13 @@ public abstract class AbstractConnectionIntegrationTests {
 		}
 	}
 
+	@Test
+	public void testLastSave() {
+		actual.add(connection.lastSave());
+		List<Object> results = getResults();
+		assertNotNull(results.get(0));
+	}
+
 	protected void verifyResults(List<Object> expected) {
 		assertEquals(expected, getResults());
 	}

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

@@ -15,11 +15,17 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+
 import org.junit.After;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.data.redis.connection.AbstractConnectionPipelineIntegrationTests;
+import org.springframework.data.redis.connection.DefaultStringRedisConnection;
 import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -53,14 +59,6 @@ public class JedisConnectionPipelineIntegrationTests extends
 	public void testGetConfig() {
 	}
 
-	@Ignore("DATAREDIS-143 MultiResponseBuilder trying to cast QUEUED to a List")
-	public void testWatch() {
-	}
-
-	@Ignore("DATAREDIS-143  MultiResponseBuilder trying to cast QUEUED to a List")
-	public void testUnwatch() {
-	}
-
 	@Ignore("https://github.com/xetorthio/jedis/pull/389 Pipeline tries to return List<String> instead of Long on sort")
 	public void testSortStore() {
 	}
@@ -73,17 +71,51 @@ public class JedisConnectionPipelineIntegrationTests extends
 	public void testSortStoreNullParams() {
 	}
 
-	@Ignore("DATAREDIS-143 Jedis ClassCastExceptions closing pipeline on certain ops")
-	public void testMultiExec() {
-	}
-
-	@Ignore("DATAREDIS-143 Jedis NPE closing pipeline on certain ops")
+	@Ignore("https://github.com/xetorthio/jedis/issues/438 Cannot get results from Pipeline after tx discard called")
 	public void testMultiDiscard() {
 	}
 
-	@Ignore("DATAREDIS-143 Jedis ClassCastExceptions closing pipeline on certain ops")
 	@Test
-	public void testErrorInTx() {
+	public void testWatch() {
+		connection.set("testitnow", "willdo");
+		connection.watch("testitnow".getBytes());
+		// Jedis doesn't actually send commands until you close the pipeline
+		getResults();
+		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(
+				connectionFactory.getConnection());
+		conn2.set("testitnow", "something");
+		conn2.close();
+		// Reopen the pipeline
+		initConnection();
+		connection.multi();
+		connection.set("testitnow", "somethingelse");
+		actual.add(connection.exec());
+		actual.add(connection.get("testitnow"));
+		verifyResults(Arrays.asList(new Object[] { null, "something" }));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testUnwatch() throws Exception {
+		connection.set("testitnow", "willdo");
+		connection.watch("testitnow".getBytes());
+		// Jedis doesn't actually send commands until you close the pipeline
+		getResults();
+		initConnection();
+		connection.unwatch();
+		// Jedis doesn't actually send commands until you close the pipeline
+		getResults();
+		initConnection();
+		connection.multi();
+		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();
+		List<Object> execResults = (List<Object>) results.get(0);
+		assertEquals("somethingelse", new String((byte[]) execResults.get(1)));
 	}
 
 	// Unsupported Ops

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

@@ -0,0 +1,80 @@
+package org.springframework.data.redis.connection.jedis;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.springframework.data.redis.connection.RedisPipelineException;
+
+public class JedisConnectionPipelineTxIntegrationTests extends JedisConnectionTransactionIntegrationTests {
+
+	@Ignore("Jedis issue: Pipeline tries to return String instead of List<String>")
+	@Test
+	public void testGetConfig() {
+	}
+
+	@Ignore("https://github.com/xetorthio/jedis/pull/389 Pipeline tries to return List<String> instead of Long on sort")
+	public void testSortStore() {
+	}
+
+	@Ignore("Jedis issue: Pipeline tries to return Long instead of List<String> on sort with no params")
+	public void testSortNullParams() {
+	}
+
+	@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() {
+	}
+
+	@Test
+	public void testEcho() {
+		actual.add(connection.echo("Hello World"));
+		verifyResults(Arrays.asList(new Object[] { "Hello World" }));
+	}
+
+	@Test(expected=RedisPipelineException.class)
+	public void exceptionExecuteNative() throws Exception {
+		connection.execute("set", "foo");
+		connection.execute("ZadD", getClass() + "#foo\t0.90\titem");
+		getResults();
+	}
+
+	@Test
+	public void testLastSave() {
+		connection.lastSave();
+		List<Object> results = getResults();
+		assertNotNull(results.get(0));
+	}
+
+	protected void initConnection() {
+		connection.openPipeline();
+		connection.multi();
+	}
+
+	@SuppressWarnings("unchecked")
+	protected List<Object> getResults() {
+		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 convertResults(txResults);
+	}
+
+	@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;
+	}
+
+}

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

@@ -135,6 +135,11 @@ public class JedisConnectionTransactionIntegrationTests extends
 		super.testBitSet();
 	}
 
+	@Test(expected = UnsupportedOperationException.class)
+	public void testBitGet() {
+		connection.getBit("bitly", 1l);
+	}
+
 	@Test(expected = UnsupportedOperationException.class)
 	public void testBitCount() {
 		connection.bitCount("foo");
@@ -212,7 +217,7 @@ public class JedisConnectionTransactionIntegrationTests extends
 
 	@Test(expected = UnsupportedOperationException.class)
 	public void testZRevRangeByScoreWithScoresOffsetCount() {
-		super.testZRevRangeByScoreWithScoresOffsetCount();
+		byteConnection.zRevRangeByScoreWithScores("myset".getBytes(), 0d, 3d, 0, 1);
 	}
 
 	@Test(expected = UnsupportedOperationException.class)
@@ -361,11 +366,21 @@ public class JedisConnectionTransactionIntegrationTests extends
 		super.testEcho();
 	}
 
+	@Test(expected = UnsupportedOperationException.class)
+	public void testSelect() {
+		super.testSelect();
+	}
+
+	@Test(expected = UnsupportedOperationException.class)
+	public void testLastSave() {
+		super.testLastSave();
+	}
+
 	@Test
 	public void exceptionExecuteNative() throws Exception {
 		actual.add(connection.execute("ZadD", getClass() + "#foo\t0.90\titem"));
 		try {
-			// Syntax error on queued commands are swallowed and no results are
+			// In Redis 2.4, syntax error on queued commands are swallowed and no results are
 			// returned
 			verifyResults(Arrays.asList(new Object[] {}));
 			if (RedisVersionUtils.atLeast("2.6.5", connection)) {

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

@@ -221,6 +221,11 @@ public class LettuceConnectionTransactionIntegrationTests extends
 		}
 	}
 
+	@Test(expected=UnsupportedOperationException.class)
+	public void testSelect() {
+		super.testSelect();
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Override
 	protected Object convertResult(Object result) {