Support zAdd with multiple values

DATAREDIS-99

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

@@ -49,6 +49,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 	private final RedisSerializer<String> serializer;
 	private Converter<byte[],String> bytesToString = new DeserializingConverter();
 	private SetConverter<Tuple,StringTuple> tupleToStringTuple = new SetConverter<Tuple,StringTuple>(new TupleConverter());
+	private SetConverter<StringTuple,Tuple> stringTupleToTuple = new SetConverter<StringTuple, Tuple>(new StringTupleConverter());
 	private ListConverter<byte[], String> byteListToStringList = new ListConverter<byte[],String>(bytesToString);
 	private MapConverter<byte[], String> byteMapToStringMap = new MapConverter<byte[],String>(bytesToString);
 	private SetConverter<byte[], String> byteSetToStringSet = new SetConverter<byte[],String>(bytesToString);
@@ -71,6 +72,12 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 		}
 	}
 
+	private class StringTupleConverter implements Converter<StringTuple, Tuple> {
+		public Tuple convert(StringTuple source) {
+			return new DefaultTuple(source.getValue(), source.getScore());
+		}
+	}
+
 	private class IdentityConverter implements Converter<Object, Object> {
 		public Object convert(Object source) {
 			return source;
@@ -905,6 +912,14 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 		return result;
 	}
 
+	public Long zAdd(byte[] key, Set<Tuple> tuples) {
+		Long result = delegate.zAdd(key, tuples);
+		if(isFutureConversion()) {
+			addResultConverter(identityConverter);
+		}
+		return result;
+	}
+
 	public Long zCard(byte[] key) {
 		Long result = delegate.zCard(key);
 		if(isFutureConversion()) {
@@ -1918,6 +1933,14 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 		return result;
 	}
 
+
+	public Long zAdd(String key, Set<StringTuple> tuples) {
+		Long result = delegate.zAdd(serialize(key), stringTupleToTuple.convert(tuples));
+		if(isFutureConversion()) {
+			addResultConverter(identityConverter);
+		}
+		return result;
+	}
 	
 	public Long zCard(String key) {
 		Long result = delegate.zCard(serialize(key));

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

@@ -44,6 +44,8 @@ public interface RedisZSetCommands {
 
 	Boolean zAdd(byte[] key, double score, byte[] value);
 
+	Long zAdd(byte[] key, Set<Tuple> tuples);
+
 	Boolean zRem(byte[] key, byte[] value);
 
 	Double zIncrBy(byte[] key, double increment, byte[] value);

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

@@ -191,6 +191,8 @@ public interface StringRedisConnection extends RedisConnection {
 
 	Boolean zAdd(String key, double score, String value);
 
+	Long zAdd(String key, Set<StringTuple> tuples);
+
 	Boolean zRem(String key, String value);
 
 	Double zIncrBy(String key, double increment, String value);

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

@@ -15,10 +15,14 @@
  */
 package org.springframework.data.redis.connection.convert;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Properties;
+import java.util.Set;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 
 /**
  * Common type converters
@@ -61,4 +65,13 @@ abstract public class Converters {
 	public static byte[] toBit(Boolean source) {
 		return (source ? ONE : ZERO);
 	}
+
+	public static List<Object> toObjects(Set<Tuple> tuples) {
+		List<Object> tupleArgs = new ArrayList<Object>(tuples.size() * 2);
+		for(Tuple tuple: tuples) {
+			tupleArgs.add(tuple.getScore());
+			tupleArgs.add(tuple.getValue());
+		}
+		return tupleArgs;
+	}
 }

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

@@ -19,6 +19,7 @@ import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -1901,6 +1902,18 @@ public class JedisConnection implements RedisConnection {
 		}
 	}
 
+	public Long zAdd(byte[] key, Set<Tuple> tuples) {
+		if(isPipelined() || isQueueing()) {
+			throw new UnsupportedOperationException("zAdd of multiple fields not supported " +
+					"in pipeline or transaction");
+		}
+		Map<Double, byte[]> args = zAddArgs(tuples);
+		try {
+			return jedis.zadd(key, args);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
 
 	public Long zCard(byte[] key) {
 		try {
@@ -2747,4 +2760,16 @@ public class JedisConnection implements RedisConnection {
 		args.add(Protocol.toByteArray(timeout));
 		return args.toArray(new byte[args.size()][]);
 	}
+
+    private Map<Double, byte[]> zAddArgs(Set<Tuple> tuples) {
+		Map<Double, byte[]> args = new HashMap<Double, byte[]>();
+		for(Tuple tuple: tuples) {
+			if(args.containsKey(tuple.getScore())) {
+				throw new UnsupportedOperationException("Bulk add of multiple elements with the same score is not supported. " +
+						"Add the elements individually.");
+			}
+			args.put(tuple.getScore(), tuple.getValue());
+		}
+		return args;
+	}
 }

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

@@ -949,6 +949,9 @@ public class JredisConnection implements RedisConnection {
 		}
 	}
 
+	public Long zAdd(byte[] key, Set<Tuple> tuples) {
+		throw new UnsupportedOperationException();
+	}
 
 	public Long zCard(byte[] key) {
 		try {

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -1977,6 +1977,25 @@ public class LettuceConnection implements RedisConnection {
 		}
 	}
 
+
+	public Long zAdd(byte[] key, Set<Tuple> tuples) {
+		try {
+			if (isPipelined()) {
+				pipeline(new LettuceResult(getAsyncConnection().zadd(key,
+						LettuceConverters.toObjects(tuples).toArray())));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(new LettuceTxResult(getConnection().zadd(key,
+						LettuceConverters.toObjects(tuples).toArray())));
+				return null;
+			}
+			return getConnection().zadd(key, LettuceConverters.toObjects(tuples).toArray());
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 	public Long zCard(byte[] key) {
 		try {
 			if (isPipelined()) {

src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java

@@ -1544,6 +1544,20 @@ public class SrpConnection implements RedisConnection {
 		}
 	}
 
+	public Long zAdd(byte[] key, Set<Tuple> tuples) {
+		try {
+			List<Object> args = new ArrayList<Object>();
+			args.add(key);
+			args.addAll(SrpConverters.toObjects(tuples));
+			if (isPipelined()) {
+				pipeline(new SrpResult(pipeline.zadd(args.toArray())));
+				return null;
+			}
+			return client.zadd(args.toArray()).data();
+		} catch (Exception ex) {
+			throw convertSrpAccessException(ex);
+		}
+	}
 
 	public Long zCard(byte[] key) {
 		try {

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

@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
@@ -199,6 +200,24 @@ abstract class AbstractOperations<K, V> {
 		return set;
 	}
 
+	@SuppressWarnings("unchecked")
+	Set<Tuple> rawTupleValues(Set<TypedTuple<V>> values) {
+		if(values == null) {
+			return null;
+		}
+		Set<Tuple> rawTuples = new LinkedHashSet<Tuple>(values.size());
+		for(TypedTuple<V> value: values) {
+			byte[] rawValue;
+			if(valueSerializer() == null && value.getValue() instanceof byte[]) {
+				rawValue = (byte[]) value.getValue();
+			}else {
+				rawValue = valueSerializer().serialize(value.getValue());
+			}
+			rawTuples.add(new DefaultTuple(rawValue, value.getScore()));
+		}
+		return rawTuples;
+	}
+
 	@SuppressWarnings("unchecked")
 	List<V> deserializeValues(List<byte[]> rawValues) {
 		if(valueSerializer() == null) {

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

@@ -61,6 +61,8 @@ public interface BoundZSetOperations<K, V> extends BoundKeyOperations<K> {
 
 	Boolean add(V value, double score);
 
+	Long add(Set<TypedTuple<V>> tuples);
+
 	Double incrementScore(V value, double delta);
 
 	Long rank(Object o);

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

@@ -47,7 +47,10 @@ class DefaultBoundZSetOperations<K, V> extends DefaultBoundKeyOperations<K> impl
 		return ops.add(getKey(), value, score);
 	}
 
-	
+	public Long add(Set<TypedTuple<V>> tuples) {
+		return ops.add(getKey(), tuples);
+	}
+
 	public Double incrementScore(V value, double delta) {
 		return ops.incrementScore(getKey(), value, delta);
 	}

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

@@ -24,7 +24,7 @@ import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
  * 
  * @author Costin Leau
  */
-class DefaultTypedTuple<V> implements TypedTuple<V> {
+public class DefaultTypedTuple<V> implements TypedTuple<V> {
 
 	private final Double score;
 	private final V value;

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

@@ -46,6 +46,17 @@ class DefaultZSetOperations<K, V> extends AbstractOperations<K, V> implements ZS
 		}, true);
 	}
 
+	public Long add(K key, Set<TypedTuple<V>> tuples) {
+		final byte[] rawKey = rawKey(key);
+		final Set<Tuple> rawValues = rawTupleValues(tuples);
+
+		return execute(new RedisCallback<Long>() {
+
+			public Long doInRedis(RedisConnection connection) {
+				return connection.zAdd(rawKey, rawValues);
+			}
+		}, true);
+	}
 
 	public Double incrementScore(K key, V value, final double delta) {
 		final byte[] rawKey = rawKey(key);

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

@@ -69,6 +69,8 @@ public interface ZSetOperations<K, V> {
 
 	Boolean add(K key, V value, double score);
 
+	Long add(K key, Set<TypedTuple<V>> tuples);
+
 	Double incrementScore(K key, V value, double delta);
 
 	Long rank(K key, Object o);