Add deserialize of exec results to RedisTemplate

DATAREDIS-228

- RedisTemplate.exec() will now deserialize raw results
using RT serializers, where applicable. This behavior
is disabled if convertPipelineAndTxResults is set
to false on the ConnectionFactory

- Add new RedisTemplate.exec(RedisSerializer) method
that will deserialize all results using the provided
serializer.

src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java

@@ -31,4 +31,17 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
 	 * @return connection for interacting with Redis.
 	 */
 	RedisConnection getConnection();
+
+	/**
+	 * Specifies if pipelined results should be converted to the expected data
+	 * type. If false, results of {@link RedisConnection#closePipeline()} and {RedisConnection#exec()}
+	 * will be of the type returned by the underlying driver
+	 *
+	 * This method is mostly for backwards compatibility with 1.0. It is generally
+	 * always a good idea to allow results to be converted and deserialized.
+	 * In fact, this is now the default behavior.
+	 *
+	 * @return Whether or not to convert pipeline and tx results
+	 */
+	boolean getConvertPipelineAndTxResults();
 }

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnectionFactory.java

@@ -201,4 +201,11 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 		Assert.isTrue(index >= 0, "invalid DB index (a positive index required)");
 		this.dbIndex = index;
 	}
+
+	/**
+	 * {@link JredisConnection} does not support pipeline or transactions
+	 */
+	public boolean getConvertPipelineAndTxResults() {
+		return false;
+	}
 }

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

@@ -120,6 +120,19 @@ public interface RedisOperations<K, V> {
 
 	List<Object> exec();
 
+	/**
+	 * Execute a transaction, using the provided {@link RedisSerializer} to deserialize
+	 * any results that are byte[]s or Collections of byte[]s. If a result is a Map, the
+	 * provided {@link RedisSerializer} will be used for both the keys and values. Other result
+	 * types (Long, Boolean, etc) are left as-is in the converted results. Tuple results are
+	 * automatically converted to TypedTuples.
+	 *
+	 * @param valueSerializer The {@link RedisSerializer} to use for deserializing the results
+	 * of transaction exec
+	 * @return The deserialized results of transaction exec
+	 */
+	List<Object> exec(RedisSerializer<?> valueSerializer);
+
 	// pubsub functionality on the template
 	void convertAndSend(String destination, Object message);
 

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

@@ -20,15 +20,20 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.connection.SortParameters;
+import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
 import org.springframework.data.redis.core.query.QueryUtils;
 import org.springframework.data.redis.core.query.SortQuery;
 import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
@@ -414,14 +419,87 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return (K) keySerializer.deserialize(value);
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private List<Object> deserializeMixedResults(List<Object> rawValues, RedisSerializer valueSerializer,
+			RedisSerializer hashKeySerializer, RedisSerializer hashValueSerializer) {
+		if(rawValues == null) {
+			return null;
+		}
+		List<Object> values = new ArrayList<Object>();
+		for(Object rawValue: rawValues) {
+			if(rawValue instanceof byte[]) {
+				values.add(valueSerializer.deserialize((byte[])rawValue));
+			} else if(rawValue instanceof List) {
+				// Lists are the only potential Collections of mixed values....
+				values.add(deserializeMixedResults((List)rawValue, valueSerializer, hashKeySerializer, hashValueSerializer));
+			} else if(rawValue instanceof Set && !(((Set)rawValue).isEmpty())) {
+				values.add(deserializeSet((Set)rawValue, valueSerializer));
+			} else if(rawValue instanceof Map && !(((Map)rawValue).isEmpty()) &&
+					((Map)rawValue).values().iterator().next() instanceof byte[]) {
+				values.add(SerializationUtils.deserialize((Map)rawValue, hashKeySerializer, hashValueSerializer));
+			} else {
+				values.add(rawValue);
+			}
+		}
+		return values;
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private Set<?> deserializeSet(Set rawSet, RedisSerializer valueSerializer) {
+		if(rawSet.isEmpty()) {
+			return rawSet;
+		}
+		Object setValue = rawSet.iterator().next();
+		if(setValue instanceof byte[]) {
+			return (SerializationUtils.deserialize((Set)rawSet, valueSerializer));
+		}else if(setValue instanceof Tuple) {
+			return deserializeTupleValues(rawSet, valueSerializer);
+		} else {
+			return rawSet;
+		}
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private Set<TypedTuple<V>> deserializeTupleValues(Set<Tuple> rawValues, RedisSerializer valueSerializer) {
+		Set<TypedTuple<V>> set = new LinkedHashSet<TypedTuple<V>>(rawValues.size());
+		for (Tuple rawValue : rawValues) {
+			set.add(new DefaultTypedTuple(valueSerializer.deserialize(rawValue.getValue()), rawValue.getScore()));
+		}
+		return set;
+	}
+
 	//
 	// RedisOperations
 	//
 
+	/**
+	 * Execute a transaction, using the default {@link RedisSerializer}s to deserialize
+	 * any results that are byte[]s or Collections or Maps of byte[]s or Tuples. Other result
+	 * types (Long, Boolean, etc) are left as-is in the converted results.
+	 *
+	 * If conversion of tx results has been disabled in the {@link ConnectionFactory},
+	 * the results of exec will be returned without deserialization. This check is mostly for
+	 * backwards compatibility with 1.0.
+	 *
+	 * @return The (possibly deserialized) results of transaction exec
+	 */
 	public List<Object> exec() {
+		List<Object> results = execRaw();
+		if(getConnectionFactory().getConvertPipelineAndTxResults()) {
+			return deserializeMixedResults(results, valueSerializer,
+					hashKeySerializer, hashValueSerializer);
+		} else {
+			return results;
+		}
+	}
+
+	public List<Object> exec(RedisSerializer<?> valueSerializer) {
+		return deserializeMixedResults(execRaw(), valueSerializer, valueSerializer,
+				valueSerializer);
+	}
+
+	protected List<Object> execRaw() {
 		return execute(new RedisCallback<List<Object>>() {
-
-
 			public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
 				return connection.exec();
 			}

src/main/java/org/springframework/data/redis/serializer/SerializationUtils.java

@@ -80,4 +80,17 @@ public abstract class SerializationUtils {
 		}
 		return ret;
 	}
+
+	public static <HK,HV> Map<HK, HV> deserialize(Map<byte[], byte[]> rawValues,
+			RedisSerializer<HK> hashKeySerializer, RedisSerializer<HV> hashValueSerializer) {
+		if (rawValues == null) {
+			return null;
+		}
+		Map<HK, HV> map = new LinkedHashMap<HK, HV>(rawValues.size());
+		for (Map.Entry<byte[], byte[]> entry : rawValues.entrySet()) {
+			map.put((HK) hashKeySerializer.deserialize(entry.getKey()),
+					(HV) hashValueSerializer.deserialize(entry.getValue()));
+		}
+		return map;
+	}
 }

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

@@ -19,16 +19,29 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.springframework.data.redis.SpinBarrier.waitFor;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.After;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.RedisTestProfileValueSource;
+import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.TestCondition;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.StringRedisConnection;
+import org.springframework.data.redis.connection.srp.SrpConnectionFactory;
+import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
+import org.springframework.data.redis.serializer.GenericToStringSerializer;
 import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.annotation.ProfileValueSourceConfiguration;
 import org.springframework.test.context.ContextConfiguration;
@@ -111,4 +124,95 @@ public class StringRedisTemplateTests {
 		});
 		assertEquals(value,"it");
 	}
+
+	@Test
+	public void testExecDeserializes() {
+		List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() {
+			@SuppressWarnings({ "rawtypes", "unchecked" })
+			public List<Object> execute(RedisOperations operations) throws DataAccessException {
+				operations.multi();
+				operations.opsForValue().set("foo", "bar");
+				// byte[]
+				operations.opsForValue().get("foo");
+				operations.opsForList().leftPush("foolist", "something");
+				// List<byte[]>
+				operations.opsForList().range("foolist", 0l, 1l);
+				operations.opsForSet().add("fooset", "a");
+				// Set<byte[]>
+				operations.opsForSet().members("fooset");
+				operations.opsForZSet().add("foozset", "Joe", 1d);
+				// Set<TypedTuple>
+				operations.opsForZSet().rangeWithScores("foozset", 0l, -1l);
+				operations.opsForHash().put("foomap", "test", "passed");
+				// Map<byte[],byte[]>
+				operations.opsForHash().entries("foomap");
+				return operations.exec();
+			}
+		});
+		List<String> list = Collections.singletonList("something");
+		Set<String> stringSet = new HashSet<String>(Collections.singletonList("a"));
+		Set<TypedTuple<String>> tupleSet = new LinkedHashSet<TypedTuple<String>>(
+				Collections.singletonList(new DefaultTypedTuple<String>("Joe", 1d)));
+		Map<String,String> map = new LinkedHashMap<String,String>();
+		map.put("test", "passed");
+		assertEquals(Arrays.asList(new Object[] {"bar", 1l, list, true, stringSet, true, tupleSet, true, map}),
+				results);
+	}
+
+	@Test
+	public void testExecCustomSerializer() {
+		List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() {
+			@SuppressWarnings({ "rawtypes", "unchecked" })
+			public List<Object> execute(RedisOperations operations) throws DataAccessException {
+				operations.multi();
+				operations.opsForValue().set("foo", "5");
+				// byte[]
+				operations.opsForValue().get("foo");
+				operations.opsForList().leftPush("foolist", "6");
+				// List<byte[]>
+				operations.opsForList().range("foolist", 0l, 1l);
+				operations.opsForSet().add("fooset", "7");
+				// Set<byte[]>
+				operations.opsForSet().members("fooset");
+				operations.opsForZSet().add("foozset", "9", 1d);
+				// Set<TypedTuple>
+				operations.opsForZSet().rangeWithScores("foozset", 0l, -1l);
+				operations.opsForZSet().range("foozset", 0, -1);
+				operations.opsForHash().put("foomap", "10", "11");
+				// Map<byte[],byte[]>
+				operations.opsForHash().entries("foomap");
+				return operations.exec(new GenericToStringSerializer<Long>(Long.class));
+			}
+		});
+		// Everything should be converted to Longs
+		List<Long> list = Collections.singletonList(6l);
+		Set<Long> longSet = new HashSet<Long>(Collections.singletonList(7l));
+		Set<TypedTuple<Long>> tupleSet = new LinkedHashSet<TypedTuple<Long>>(
+				Collections.singletonList(new DefaultTypedTuple<Long>(9l, 1d)));
+		Set<Long> zSet = new LinkedHashSet<Long>(Collections.singletonList(9l));
+		Map<Long, Long> map = new LinkedHashMap<Long,Long>();
+		map.put(10l, 11l);
+		assertEquals(Arrays.asList(new Object[] {5l, 1l, list, true, longSet, true, tupleSet, zSet, true, map}),
+				results);
+	}
+
+	@Test
+	public void testExecConversionDisabled() {
+		SrpConnectionFactory factory2 = new SrpConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort());
+		factory2.setConvertPipelineAndTxResults(false);
+		factory2.afterPropertiesSet();
+		StringRedisTemplate template = new StringRedisTemplate(factory2);
+		template.afterPropertiesSet();
+		List<Object> results = template.execute(new SessionCallback<List<Object>>() {
+			@SuppressWarnings({ "rawtypes", "unchecked" })
+			public List<Object> execute(RedisOperations operations) throws DataAccessException {
+				operations.multi();
+				operations.opsForValue().set("foo","bar");
+				operations.opsForValue().get("foo");
+				return operations.exec();
+			}
+		});
+		// first value is "OK" from set call, results should still be in byte[]
+		assertEquals("bar", new String((byte[])results.get(1)));
+	}
 }