Convert Jedis pipelined results to correct data type

DATAREDIS-200

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

@@ -47,12 +47,11 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	private int port = Protocol.DEFAULT_PORT;
 	private int timeout = Protocol.DEFAULT_TIMEOUT;
 	private String password;
-
 	private boolean usePool = true;
 	private JedisPool pool = null;
 	private JedisPoolConfig poolConfig = new JedisPoolConfig();
-
 	private int dbIndex = 0;
+	private boolean convertPipelineResults = true;
 
 	/**
 	 * Constructs a new <code>JedisConnectionFactory</code> instance
@@ -146,7 +145,10 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 
 	public JedisConnection getConnection() {
 		Jedis jedis = fetchJedisConnector();
-		return postProcessConnection((usePool ? new JedisConnection(jedis, pool, dbIndex) : new JedisConnection(jedis, null, dbIndex)));
+		JedisConnection connection = (usePool ? new JedisConnection(jedis, pool, dbIndex) :
+			new JedisConnection(jedis, null, dbIndex));
+		connection.setConvertPipelineResults(convertPipelineResults);
+		return postProcessConnection(connection);
 	}
 
 
@@ -299,4 +301,26 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		Assert.isTrue(index >= 0, "invalid DB index (a positive index required)");
 		this.dbIndex = index;
 	}
+
+	/**
+	 * 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
+	 *
+	 * @return Whether or not to convert pipeline results
+	 */
+	public boolean getConvertPipelineResults() {
+		return convertPipelineResults;
+	}
+
+	/**
+	 * 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
+	 *
+	 * @param convertPipelineResults Whether or not to convert pipeline results
+	 */
+	public void setConvertPipelineResults(boolean convertPipelineResults) {
+		this.convertPipelineResults = convertPipelineResults;
+	}
 }

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

@@ -0,0 +1,180 @@
+/*
+ * 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 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.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;
+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;
+
+/**
+ * Jedis type converters
+ *
+ * @author Jennifer Hickey
+ *
+ */
+abstract public class JedisConverters extends Converters {
+
+	private static final Converter<String, byte[]> STRING_TO_BYTES;
+	private static final ListConverter<String, byte[]> STRING_LIST_TO_BYTE_LIST;
+	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;
+
+	static {
+		STRING_TO_BYTES = new Converter<String, byte[]>() {
+			public byte[] convert(String source) {
+				return source == null ? null : SafeEncoder.encode(source);
+			}
+		};
+		STRING_LIST_TO_BYTE_LIST = new ListConverter<String, byte[]>(STRING_TO_BYTES);
+		STRING_SET_TO_BYTE_SET = new SetConverter<String, byte[]>(STRING_TO_BYTES);
+		STRING_MAP_TO_BYTE_MAP = new MapConverter<String, byte[]>(STRING_TO_BYTES);
+		TUPLE_SET_TO_TUPLE_SET = new SetConverter<redis.clients.jedis.Tuple, Tuple>(
+				new Converter<redis.clients.jedis.Tuple, Tuple>() {
+					public Tuple convert(redis.clients.jedis.Tuple source) {
+						return new DefaultTuple(source.getBinaryElement(), source.getScore());
+					}
+
+				});
+	}
+
+	public static Converter<String, byte[]> stringToBytes() {
+		return STRING_TO_BYTES;
+	}
+
+	public static ListConverter<String, byte[]> stringListToByteList() {
+		return STRING_LIST_TO_BYTE_LIST;
+	}
+
+	public static SetConverter<String, byte[]> stringSetToByteSet() {
+		return STRING_SET_TO_BYTE_SET;
+	}
+
+	public static MapConverter<String, byte[]> stringMapToByteMap() {
+		return STRING_MAP_TO_BYTE_MAP;
+	}
+
+	public static SetConverter<redis.clients.jedis.Tuple, Tuple> tupleSetToTupleSet() {
+		return TUPLE_SET_TO_TUPLE_SET;
+	}
+
+	public static String[] toStrings(byte[][] source) {
+		String[] result = new String[source.length];
+		for (int i = 0; i < source.length; i++) {
+			result[i] = SafeEncoder.encode(source[i]);
+		}
+		return result;
+	}
+
+	public static Set<Tuple> toTupleSet(Set<redis.clients.jedis.Tuple> source) {
+		return TUPLE_SET_TO_TUPLE_SET.convert(source);
+	}
+
+	public static byte[] toBytes(Integer source) {
+		return String.valueOf(source).getBytes();
+	}
+
+	public static String toString(byte[] source) {
+		return source == null ? null : SafeEncoder.encode(source);
+	}
+
+	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);
+	}
+
+	public static LIST_POSITION toListPosition(Position source) {
+		Assert.notNull("list positions are mandatory");
+		return (Position.AFTER.equals(source) ? LIST_POSITION.AFTER : LIST_POSITION.BEFORE);
+	}
+
+	public static byte[][] toByteArrays(Map<byte[], byte[]> source) {
+		byte[][] result = new byte[source.size() * 2][];
+		int index = 0;
+		for (Map.Entry<byte[], byte[]> entry : source.entrySet()) {
+			result[index++] = entry.getKey();
+			result[index++] = entry.getValue();
+		}
+		return result;
+	}
+
+	public static SortingParams toSortingParams(SortParameters params) {
+		SortingParams jedisParams = null;
+		if (params != null) {
+			jedisParams = new SortingParams();
+			byte[] byPattern = params.getByPattern();
+			if (byPattern != null) {
+				jedisParams.by(params.getByPattern());
+			}
+			byte[][] getPattern = params.getGetPattern();
+			if (getPattern != null) {
+				jedisParams.get(getPattern);
+			}
+			Range limit = params.getLimit();
+			if (limit != null) {
+				jedisParams.limit((int) limit.getStart(), (int) limit.getCount());
+			}
+			Order order = params.getOrder();
+			if (order != null && order.equals(Order.DESC)) {
+				jedisParams.desc();
+			}
+			Boolean isAlpha = params.isAlphabetic();
+			if (isAlpha != null && isAlpha) {
+				jedisParams.alpha();
+			}
+		}
+		return jedisParams;
+	}
+}

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

@@ -0,0 +1,73 @@
+/*
+ * 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.util.ArrayList;
+import java.util.List;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.redis.connection.ReturnType;
+
+import redis.clients.util.SafeEncoder;
+
+/**
+ * Converts the value returned by Jedis script eval to the expected
+ * {@link ReturnType}
+ *
+ * @author Jennifer Hickey
+ *
+ */
+public class JedisScriptReturnConverter implements Converter<Object, Object> {
+
+	private final ReturnType returnType;
+
+	public JedisScriptReturnConverter(ReturnType returnType) {
+		this.returnType = returnType;
+	}
+
+	@SuppressWarnings("unchecked")
+	public Object convert(Object result) {
+		if (result instanceof String) {
+			// evalsha converts byte[] to String. Convert back for consistency
+			return SafeEncoder.encode((String) result);
+		}
+		if (returnType == ReturnType.STATUS) {
+			return JedisConverters.toString((byte[]) result);
+		}
+		if (returnType == ReturnType.BOOLEAN) {
+			// Lua false comes back as a null bulk reply
+			if (result == null) {
+				return Boolean.FALSE;
+			}
+			return ((Long) result == 1);
+		}
+		if (returnType == ReturnType.MULTI) {
+			List<Object> resultList = (List<Object>) result;
+			List<Object> convertedResults = new ArrayList<Object>();
+			for (Object res : resultList) {
+				if (res instanceof String) {
+					// evalsha converts byte[] to String. Convert back for
+					// consistency
+					convertedResults.add(SafeEncoder.encode((String) res));
+				} else {
+					convertedResults.add(res);
+				}
+			}
+			return convertedResults;
+		}
+		return result;
+	}
+}