Make Jedis return converted tx results

DATAREDIS-208

src/main/java/org/springframework/data/redis/connection/convert/TransactionResultConverter.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection.convert;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.FutureResult;
+
+/**
+ * Converts the results of transaction exec using a supplied Queue of {@link FutureResult}s.
+ * Converts any Exception objects returned in the list as well, using the supplied Exception
+ * {@link Converter}
+ *
+ * @author Jennifer Hickey
+ *
+ * @param <T>
+ *            The type of {@link FutureResult} of the individual tx operations
+ */
+public class TransactionResultConverter<T> implements Converter<List<Object>, List<Object>> {
+
+	private Queue<FutureResult<T>> txResults = new LinkedList<FutureResult<T>>();
+
+	private Converter<Exception, DataAccessException> exceptionConverter;
+
+	public TransactionResultConverter(Queue<FutureResult<T>> txResults,
+			Converter<Exception, DataAccessException> exceptionConverter) {
+		this.txResults = txResults;
+		this.exceptionConverter = exceptionConverter;
+	}
+
+	public List<Object> convert(List<Object> execResults) {
+		if (execResults == null) {
+			return null;
+		}
+		if (execResults.size() != txResults.size()) {
+			throw new IllegalArgumentException(
+					"Incorrect number of transaction results. Expected: " + txResults.size()
+							+ " Actual: " + execResults.size());
+		}
+		List<Object> convertedResults = new ArrayList<Object>();
+		for (Object result : execResults) {
+			FutureResult<T> futureResult = txResults.remove();
+			if (result instanceof Exception) {
+				throw exceptionConverter.convert((Exception) result);
+			}
+			if (!(futureResult.isStatus())) {
+				convertedResults.add(futureResult.convert(result));
+			}
+		}
+		return convertedResults;
+	}
+}

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -192,8 +192,7 @@ public class JedisConnection implements RedisConnection {
 				});
 				if(isPipelined()) {
 					pipeline(new JedisResult(result));
-				}
-				if(isQueueing()) {
+				}else {
 					transaction(new JedisResult(result));
 				}
 				return null;
@@ -314,6 +313,11 @@ public class JedisConnection implements RedisConnection {
 					cause = dataAccessException;
 				}
 				results.add(dataAccessException);
+			}catch(DataAccessException e) {
+				if (cause == null) {
+					cause = e;
+				}
+				results.add(e);
 			}
 		}
 		if (cause != null) {
@@ -425,7 +429,7 @@ public class JedisConnection implements RedisConnection {
 				return;
 			}
 			if (isQueueing()) {
-				transaction(new JedisResult(transaction.flushDB()));
+				transaction(new JedisStatusResult(transaction.flushDB()));
 				return;
 			}
 			jedis.flushDB();
@@ -699,11 +703,13 @@ public class JedisConnection implements RedisConnection {
 		try {
 			if (isPipelined()) {
 				pipeline(new JedisResult(pipeline.exec(),
-						new TransactionResultConverter<Response<?>>(new LinkedList<FutureResult<Response<?>>>(txResults))));
+						new TransactionResultConverter<Response<?>>(new LinkedList<FutureResult<Response<?>>>(txResults),
+								JedisConverters.exceptionConverter())));
 				return null;
 			}
 			List<Object> results = transaction.exec();
-			return convertPipelineAndTxResults ? new TransactionResultConverter<Response<?>>(txResults).convert(results) : results;
+			return convertPipelineAndTxResults ? new TransactionResultConverter<Response<?>>(txResults,
+					JedisConverters.exceptionConverter()).convert(results) : results;
 		} catch (Exception ex) {
 			throw convertJedisAccessException(ex);
 		} finally {

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java

@@ -51,7 +51,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	private JedisPool pool = null;
 	private JedisPoolConfig poolConfig = new JedisPoolConfig();
 	private int dbIndex = 0;
-	private boolean convertPipelineResults = true;
+	private boolean convertPipelineAndTxResults = true;
 
 	/**
 	 * Constructs a new <code>JedisConnectionFactory</code> instance
@@ -147,7 +147,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		Jedis jedis = fetchJedisConnector();
 		JedisConnection connection = (usePool ? new JedisConnection(jedis, pool, dbIndex) :
 			new JedisConnection(jedis, null, dbIndex));
-		connection.setConvertPipelineResults(convertPipelineResults);
+		connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults);
 		return postProcessConnection(connection);
 	}
 
@@ -304,23 +304,23 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 
 	/**
 	 * Specifies if pipelined results should be converted to the expected data
-	 * type. If false, results of {@link #closePipeline()} will be of the
-	 * type returned by the Jedis driver
+	 * type. If false, results of {@link JedisConnection#closePipeline()} and
+	 * {@link JedisConnection#exec()} will be of the type returned by the Jedis driver
 	 *
-	 * @return Whether or not to convert pipeline results
+	 * @return Whether or not to convert pipeline and tx results
 	 */
-	public boolean getConvertPipelineResults() {
-		return convertPipelineResults;
+	public boolean getConvertPipelineAndTxResults() {
+		return convertPipelineAndTxResults;
 	}
 
 	/**
 	 * Specifies if pipelined results should be converted to the expected data
-	 * type. If false, results of {@link #closePipeline()} will be of the
-	 * type returned by the Jedis driver
+	 * type. If false, results of {@link JedisConnection#closePipeline()} and
+	 * {@link JedisConnection#exec()} will be of the type returned by the Jedis driver
 	 *
-	 * @param convertPipelineResults Whether or not to convert pipeline results
+	 * @param convertPipelineAndTxResults Whether or not to convert pipeline and tx results
 	 */
-	public void setConvertPipelineResults(boolean convertPipelineResults) {
-		this.convertPipelineResults = convertPipelineResults;
+	public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) {
+		this.convertPipelineAndTxResults = convertPipelineAndTxResults;
 	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java

@@ -15,22 +15,17 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
-import java.io.IOException;
-import java.net.UnknownHostException;
 import java.util.Map;
 import java.util.Set;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
-import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.data.redis.RedisConnectionFailureException;
-import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
 import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
+import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.SortParameters.Order;
 import org.springframework.data.redis.connection.SortParameters.Range;
-import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.connection.convert.ListConverter;
 import org.springframework.data.redis.connection.convert.MapConverter;
@@ -38,9 +33,6 @@ import org.springframework.data.redis.connection.convert.SetConverter;
 import org.springframework.util.Assert;
 
 import redis.clients.jedis.BinaryClient.LIST_POSITION;
-import redis.clients.jedis.exceptions.JedisConnectionException;
-import redis.clients.jedis.exceptions.JedisDataException;
-import redis.clients.jedis.exceptions.JedisException;
 import redis.clients.jedis.SortingParams;
 import redis.clients.util.SafeEncoder;
 
@@ -57,6 +49,7 @@ abstract public class JedisConverters extends Converters {
 	private static final SetConverter<String, byte[]> STRING_SET_TO_BYTE_SET;
 	private static final MapConverter<String, byte[]> STRING_MAP_TO_BYTE_MAP;
 	private static final SetConverter<redis.clients.jedis.Tuple, Tuple> TUPLE_SET_TO_TUPLE_SET;
+	private static final Converter<Exception,DataAccessException> EXCEPTION_CONVERTER = new JedisExceptionConverter();
 
 	static {
 		STRING_TO_BYTES = new Converter<String, byte[]>() {
@@ -96,6 +89,10 @@ abstract public class JedisConverters extends Converters {
 		return TUPLE_SET_TO_TUPLE_SET;
 	}
 
+	public static Converter<Exception,DataAccessException> exceptionConverter() {
+		return EXCEPTION_CONVERTER;
+	}
+
 	public static String[] toStrings(byte[][] source) {
 		String[] result = new String[source.length];
 		for (int i = 0; i < source.length; i++) {
@@ -117,22 +114,7 @@ abstract public class JedisConverters extends Converters {
 	}
 
 	public static DataAccessException toDataAccessException(Exception ex) {
-		if (ex instanceof JedisDataException) {
-			return new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
-		}
-		if (ex instanceof JedisConnectionException) {
-			return new RedisConnectionFailureException(ex.getMessage(), ex);
-		}
-		if (ex instanceof JedisException) {
-			return new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
-		}
-		if (ex instanceof UnknownHostException) {
-			return new RedisConnectionFailureException("Unknown host " + ex.getMessage(), ex);
-		}
-		if (ex instanceof IOException) {
-			return new RedisConnectionFailureException("Could not connect to Redis server", ex);
-		}
-		return new RedisSystemException("Unknown jedis exception", ex);
+		return EXCEPTION_CONVERTER.convert(ex);
 	}
 
 	public static LIST_POSITION toListPosition(Position source) {

src/main/java/org/springframework/data/redis/connection/jedis/JedisExceptionConverter.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection.jedis;
+
+import java.io.IOException;
+import java.net.UnknownHostException;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.RedisSystemException;
+
+import redis.clients.jedis.exceptions.JedisConnectionException;
+import redis.clients.jedis.exceptions.JedisDataException;
+import redis.clients.jedis.exceptions.JedisException;
+
+/**
+ * Converts Exceptions thrown from Jedis to {@link DataAccessException}s
+ *
+ * @author Jennifer Hickey
+ *
+ */
+public class JedisExceptionConverter implements Converter<Exception, DataAccessException> {
+
+	public DataAccessException convert(Exception ex) {
+		if (ex instanceof DataAccessException) {
+			return (DataAccessException) ex;
+		}
+		if (ex instanceof JedisDataException) {
+			return new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
+		}
+		if (ex instanceof JedisConnectionException) {
+			return new RedisConnectionFailureException(ex.getMessage(), ex);
+		}
+		if (ex instanceof JedisException) {
+			return new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
+		}
+		if (ex instanceof UnknownHostException) {
+			return new RedisConnectionFailureException("Unknown host " + ex.getMessage(), ex);
+		}
+		if (ex instanceof IOException) {
+			return new RedisConnectionFailureException("Could not connect to Redis server", ex);
+		}
+		return new RedisSystemException("Unknown jedis exception", ex);
+	}
+}

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

@@ -685,8 +685,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		actual.add(connection.exec());
 		List<Object> results = getResults();
 		List<Object> execResults = (List<Object>) results.get(0);
-		assertEquals(2, execResults.size());
-		assertEquals("value", new String((byte[]) execResults.get(1)));
+		assertEquals(Arrays.asList(new Object[] {"value"}), execResults);
 		assertEquals("value", connection.get("key"));
 	}
 
@@ -763,7 +762,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		actual.add(connection.exec());
 		List<Object> results = getResults();
 		List<Object> execResults = (List<Object>) results.get(0);
-		assertEquals("somethingelse", new String((byte[]) execResults.get(1)));
+		assertEquals(Arrays.asList(new Object[] {"somethingelse"}), execResults);
 	}
 
 	@Test

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

@@ -0,0 +1,116 @@
+package org.springframework.data.redis.connection;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class ConvertingAbstractConnectionTransactionIntegrationTests extends
+		AbstractConnectionIntegrationTests {
+
+	@Ignore
+	public void testMultiDiscard() {
+	}
+
+	@Ignore
+	public void testMultiExec() {
+	}
+
+	@Ignore
+	public void testUnwatch() {
+	}
+
+	@Ignore
+	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
+	 * 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 {
+	}
+
+	@Test(expected = UnsupportedOperationException.class)
+	public void testWatchWhileInTx() {
+		connection.watch("foo".getBytes());
+	}
+
+	protected void initConnection() {
+		connection.multi();
+	}
+
+	protected List<Object> getResults() {
+		return connection.exec();
+	}
+
+	protected void verifyResults(List<Object> expected) {
+		List<Object> expectedTx = new ArrayList<Object>();
+		for (int i = 0; i < actual.size(); i++) {
+			expectedTx.add(null);
+		}
+		assertEquals(expectedTx, actual);
+		List<Object> results = getResults();
+		assertEquals(expected, results);
+	}
+}

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

@@ -16,10 +16,6 @@
 
 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;
@@ -30,8 +26,6 @@ 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}
  *
@@ -57,18 +51,6 @@ 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();
@@ -231,4 +213,9 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 	public void testExecWithoutMulti() {
 		super.testExecWithoutMulti();
 	}
+
+	@Test(expected=InvalidDataAccessApiUsageException.class)
+	public void testErrorInTx() {
+		super.testErrorInTx();
+	}
 }

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

@@ -115,7 +115,7 @@ public class JedisConnectionPipelineIntegrationTests extends
 		actual.add(connection.exec());
 		List<Object> results = getResults();
 		List<Object> execResults = (List<Object>) results.get(0);
-		assertEquals("somethingelse", new String((byte[]) execResults.get(1)));
+		assertEquals(Arrays.asList(new Object[] {"somethingelse"}), execResults);
 	}
 
 	// Unsupported Ops

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

@@ -57,17 +57,6 @@ public class JedisConnectionPipelineTxIntegrationTests extends JedisConnectionTr
 
 	@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
@@ -76,5 +65,4 @@ public class JedisConnectionPipelineTxIntegrationTests extends JedisConnectionTr
 		// 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

@@ -15,28 +15,15 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
-import static org.junit.Assert.fail;
-
-import java.util.Arrays;
-import java.util.LinkedHashSet;
-import java.util.Set;
-
 import org.junit.After;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.springframework.dao.DataAccessException;
-import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.data.redis.RedisVersionUtils;
-import org.springframework.data.redis.connection.AbstractConnectionTransactionIntegrationTests;
-import org.springframework.data.redis.connection.DefaultStringTuple;
-import org.springframework.data.redis.connection.StringRedisConnection.StringTuple;
+import org.springframework.data.redis.connection.ConvertingAbstractConnectionTransactionIntegrationTests;
 import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import redis.clients.jedis.Tuple;
-
 /**
  * Integration test of {@link JedisConnection} transaction functionality.
  * <p>
@@ -50,7 +37,7 @@ import redis.clients.jedis.Tuple;
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration("JedisConnectionIntegrationTests-context.xml")
 public class JedisConnectionTransactionIntegrationTests extends
-		AbstractConnectionTransactionIntegrationTests {
+		ConvertingAbstractConnectionTransactionIntegrationTests {
 
 	@After
 	public void tearDown() {
@@ -375,41 +362,4 @@ public class JedisConnectionTransactionIntegrationTests extends
 	public void testLastSave() {
 		super.testLastSave();
 	}
-
-	@Test
-	public void exceptionExecuteNative() throws Exception {
-		actual.add(connection.execute("ZadD", getClass() + "#foo\t0.90\titem"));
-		try {
-			// 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)) {
-				fail("Redis 2.6 should throw an Exception on exec if commands are invalid");
-			}
-		} catch (InvalidDataAccessApiUsageException e) {
-		}
-	}
-
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	protected Object convertResult(Object result) {
-		Object convertedResult = super.convertResult(result);
-		if (convertedResult instanceof Set) {
-		    if (!(((Set) convertedResult).isEmpty())
-					&& ((Set) convertedResult).iterator().next() instanceof Tuple) {
-				Set<StringTuple> tuples = new LinkedHashSet<StringTuple>();
-				for (Tuple value : ((Set<Tuple>) convertedResult)) {
-					DefaultStringTuple tuple = new DefaultStringTuple(
-							(byte[]) value.getBinaryElement(), value.getElement(), value.getScore());
-					tuples.add(tuple);
-				}
-				return tuples;
-			}
-		}
-		return convertedResult;
-	}
-
-	@Override
-	protected DataAccessException convertException(Exception ex) {
-		return JedisConverters.toDataAccessException(ex);
-	}
 }