+ refactored RedisTemplate by breaking it into multiple classes

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/AbstractOperations.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+import org.springframework.data.keyvalue.redis.serializer.RedisSerializer;
+import org.springframework.util.Assert;
+
+/**
+ * Internal base class used by various RedisTemplate XXXOperations implementations.
+ * 
+ * @author Costin Leau
+ */
+abstract class AbstractOperations<K, V> {
+
+	// utility methods for the template internal methods
+	abstract class ValueDeserializingRedisCallback implements RedisCallback<V> {
+		private Object key;
+
+		public ValueDeserializingRedisCallback(Object key) {
+			this.key = key;
+		}
+
+		@Override
+		public final V doInRedis(RedisConnection connection) {
+			byte[] result = inRedis(rawKey(key), connection);
+			return deserializeValue(result);
+		}
+
+		protected abstract byte[] inRedis(byte[] rawKey, RedisConnection connection);
+	}
+
+	RedisSerializer keySerializer = null;
+	RedisSerializer valueSerializer = null;
+	RedisSerializer hashKeySerializer = null;
+	RedisSerializer hashValueSerializer = null;
+	RedisSerializer stringSerializer = null;
+	RedisTemplate<K, V> template;
+
+	AbstractOperations(RedisTemplate<K, V> template) {
+		keySerializer = template.getKeySerializer();
+		valueSerializer = template.getValueSerializer();
+		hashKeySerializer = template.getHashKeySerializer();
+		hashValueSerializer = template.getHashValueSerializer();
+		stringSerializer = template.getStringSerializer();
+
+		this.template = template;
+	}
+
+
+	<T> T execute(RedisCallback<T> callback, boolean b) {
+		return template.execute(callback, b);
+	}
+
+	public RedisOperations<K, V> getOperations() {
+		return template;
+	}
+
+	@SuppressWarnings("unchecked")
+	byte[] rawKey(Object key) {
+		Assert.notNull(key, "non null key required");
+		return keySerializer.serialize(key);
+	}
+
+	byte[] rawString(String key) {
+		return stringSerializer.serialize(key);
+	}
+
+	@SuppressWarnings("unchecked")
+	byte[] rawValue(Object value) {
+		return valueSerializer.serialize(value);
+	}
+
+	@SuppressWarnings("unchecked")
+	<HK> byte[] rawHashKey(HK hashKey) {
+		Assert.notNull(hashKey, "non null hash key required");
+		return hashKeySerializer.serialize(hashKey);
+	}
+
+	@SuppressWarnings("unchecked")
+	<HV> byte[] rawHashValue(HV value) {
+		return hashValueSerializer.serialize(value);
+	}
+
+	byte[][] rawKeys(K key, K otherKey) {
+		final byte[][] rawKeys = new byte[2][];
+
+
+		rawKeys[0] = rawKey(key);
+		rawKeys[1] = rawKey(key);
+		return rawKeys;
+	}
+
+	byte[][] rawKeys(Collection<K> keys) {
+		return rawKeys(null, keys);
+	}
+
+	byte[][] rawKeys(K key, Collection<K> keys) {
+		final byte[][] rawKeys = new byte[keys.size() + (key != null ? 1 : 0)][];
+
+		int i = 0;
+
+		if (key != null) {
+			rawKeys[i++] = rawKey(key);
+		}
+
+		for (K k : keys) {
+			rawKeys[i++] = rawKey(k);
+		}
+
+		return rawKeys;
+	}
+
+	<T extends Collection<V>> T deserializeValues(Collection<byte[]> rawValues, Class<T> type) {
+		return SerializationUtils.deserializeValues(rawValues, type, valueSerializer);
+	}
+
+	@SuppressWarnings("unchecked")
+	<T> Set<T> deserializeHashKeys(Collection<byte[]> rawKeys) {
+		return SerializationUtils.deserializeValues(rawKeys, Set.class, hashKeySerializer);
+	}
+
+	@SuppressWarnings("unchecked")
+	<T> List<T> deserializeHashValues(Collection<byte[]> rawValues) {
+		return SerializationUtils.deserializeValues(rawValues, List.class, hashValueSerializer);
+	}
+
+	@SuppressWarnings("unchecked")
+	<HK, HV> Map<HK, HV> deserializeHashMap(Map<byte[], byte[]> entries) {
+		Map<HK, HV> map = new LinkedHashMap<HK, HV>(entries.size());
+
+		for (Map.Entry<byte[], byte[]> entry : entries.entrySet()) {
+			map.put((HK) deserializeHashKey(entry.getKey()), (HV) deserializeHashValue(entry.getValue()));
+		}
+
+		return map;
+	}
+
+	@SuppressWarnings("unchecked")
+	K deserializeKey(byte[] value) {
+		return (K) SerializationUtils.deserialize(value, keySerializer);
+	}
+
+	@SuppressWarnings("unchecked")
+	V deserializeValue(byte[] value) {
+		return (V) SerializationUtils.deserialize(value, valueSerializer);
+	}
+
+	@SuppressWarnings("unchecked")
+	String deserializeString(byte[] value) {
+		return (String) SerializationUtils.deserialize(value, stringSerializer);
+	}
+
+	@SuppressWarnings( { "unchecked" })
+	<HK> HK deserializeHashKey(byte[] value) {
+		return (HK) SerializationUtils.deserialize(value, hashKeySerializer);
+	}
+
+	@SuppressWarnings("unchecked")
+	<HV> HV deserializeHashValue(byte[] value) {
+		return (HV) SerializationUtils.deserialize(value, hashValueSerializer);
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/CloseSuppressingInvocationHandler.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+
+/**
+* Invocation handler that suppresses close calls on {@link RedisConnection}.
+* @see RedisConnection#close()
+* @author Costin Leau
+*/
+class CloseSuppressingInvocationHandler implements InvocationHandler {
+
+	private static final String CLOSE = "close";
+	private static final String HASH_CODE = "hashCode";
+	private static final String EQUALS = "equals";
+
+	private final RedisConnection target;
+
+	public CloseSuppressingInvocationHandler(RedisConnection target) {
+		this.target = target;
+	}
+
+	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+
+		if (method.getName().equals(EQUALS)) {
+			// Only consider equal when proxies are identical.
+			return (proxy == args[0]);
+		}
+		else if (method.getName().equals(HASH_CODE)) {
+			// Use hashCode of PersistenceManager proxy.
+			return System.identityHashCode(proxy);
+		}
+		else if (method.getName().equals(CLOSE)) {
+			// Handle close method: suppress, not valid.
+			return null;
+		}
+
+		// Invoke method on target RedisConnection.
+		try {
+			Object retVal = method.invoke(this.target, args);
+			return retVal;
+		} catch (InvocationTargetException ex) {
+			throw ex.getTargetException();
+		}
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultHashOperations.java

@@ -0,0 +1,228 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+
+/**
+ * Default implementation of {@link HashOperations}.
+ *  
+ * @author Costin Leau
+ */
+class DefaultHashOperations<K, HK, HV> extends AbstractOperations<K, Object> implements HashOperations<K, HK, HV> {
+
+	@SuppressWarnings("unchecked")
+	DefaultHashOperations(RedisTemplate<K, ?> template) {
+		super((RedisTemplate<K, Object>) template);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public HV get(K key, Object hashKey) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawHashKey = rawHashKey(hashKey);
+
+		byte[] rawHashValue = execute(new RedisCallback<byte[]>() {
+			@Override
+			public byte[] doInRedis(RedisConnection connection) {
+				return connection.hGet(rawKey, rawHashKey);
+			}
+		}, true);
+
+		return (HV) deserializeHashValue(rawHashValue);
+	}
+
+	@Override
+	public Boolean hasKey(K key, Object hashKey) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawHashKey = rawHashKey(hashKey);
+
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.hExists(rawKey, rawHashKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long increment(K key, HK hashKey, final long delta) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawHashKey = rawHashKey(hashKey);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.hIncrBy(rawKey, rawHashKey, delta);
+			}
+		}, true);
+
+	}
+
+	@Override
+	public Set<HK> keys(K key) {
+		final byte[] rawKey = rawKey(key);
+
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.hKeys(rawKey);
+			}
+		}, true);
+
+		return deserializeHashKeys(rawValues);
+	}
+
+	@Override
+	public Long size(K key) {
+		final byte[] rawKey = rawKey(key);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.hLen(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public void putAll(K key, Map<? extends HK, ? extends HV> m) {
+		if (m.isEmpty()) {
+			return;
+		}
+
+		final byte[] rawKey = rawKey(key);
+
+		final Map<byte[], byte[]> hashes = new LinkedHashMap<byte[], byte[]>(m.size());
+
+		for (Map.Entry<? extends HK, ? extends HV> entry : m.entrySet()) {
+			hashes.put(rawHashKey(entry.getKey()), rawHashValue(entry.getValue()));
+		}
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.hMSet(rawKey, hashes);
+				return null;
+			}
+		}, true);
+	}
+
+
+	@Override
+	public Collection<HV> multiGet(K key, Collection<HK> fields) {
+		if (fields.isEmpty()) {
+			return Collections.emptyList();
+		}
+
+		final byte[] rawKey = rawKey(key);
+
+		final byte[][] rawHashKeys = new byte[fields.size()][];
+
+		int counter = 0;
+		for (HK hashKey : fields) {
+			rawHashKeys[counter++] = rawHashKey(hashKey);
+		}
+
+		List<byte[]> rawValues = execute(new RedisCallback<List<byte[]>>() {
+			@Override
+			public List<byte[]> doInRedis(RedisConnection connection) {
+				return connection.hMGet(rawKey, rawHashKeys);
+			}
+		}, true);
+
+		return deserializeHashValues(rawValues);
+	}
+
+	@Override
+	public void put(K key, HK hashKey, HV value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawHashKey = rawHashKey(hashKey);
+		final byte[] rawHashValue = rawHashValue(value);
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.hSet(rawKey, rawHashKey, rawHashValue);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public Boolean putIfAbsent(K key, HK hashKey, HV value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawHashKey = rawHashKey(hashKey);
+		final byte[] rawHashValue = rawHashValue(value);
+
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.hSetNX(rawKey, rawHashKey, rawHashValue);
+			}
+		}, true);
+	}
+
+
+	@Override
+	public List<HV> values(K key) {
+		final byte[] rawKey = rawKey(key);
+
+		List<byte[]> rawValues = execute(new RedisCallback<List<byte[]>>() {
+			@Override
+			public List<byte[]> doInRedis(RedisConnection connection) {
+				return connection.hVals(rawKey);
+			}
+		}, true);
+
+		return deserializeHashValues(rawValues);
+	}
+
+	@Override
+	public void delete(K key, Object hashKey) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawHashKey = rawHashKey(hashKey);
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.hDel(rawKey, rawHashKey);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public Map<HK, HV> entries(K key) {
+		final byte[] rawKey = rawKey(key);
+
+		Map<byte[], byte[]> entries = execute(new RedisCallback<Map<byte[], byte[]>>() {
+			@Override
+			public Map<byte[], byte[]> doInRedis(RedisConnection connection) {
+				return connection.hGetAll(rawKey);
+			}
+		}, true);
+
+		return deserializeHashMap(entries);
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultListOperations.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+import org.springframework.data.keyvalue.redis.connection.RedisListCommands.Position;
+
+/**
+ * Default implementation of {@link ListOperations}.
+ * 
+ * @author Costin Leau
+ */
+class DefaultListOperations<K, V> extends AbstractOperations<K, V> implements ListOperations<K, V> {
+
+	DefaultListOperations(RedisTemplate<K, V> template) {
+		super(template);
+	}
+
+	@Override
+	public V index(K key, final long index) {
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.lIndex(rawKey, index);
+			}
+		}, true);
+	}
+
+	@Override
+	public V leftPop(K key) {
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.lPop(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public V leftPop(K key, long timeout, TimeUnit unit) {
+		final int tm = (int) unit.toSeconds(timeout);
+
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.bLPop(tm, rawKey).get(0);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long leftPush(K key, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.lPush(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long leftPushIfPresent(K key, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.lPushX(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long leftPush(K key, V pivot, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawPivot = rawValue(pivot);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.lInsert(rawKey, Position.BEFORE, rawPivot, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long size(K key) {
+		final byte[] rawKey = rawKey(key);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.lLen(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public List<V> range(K key, final long start, final long end) {
+		final byte[] rawKey = rawKey(key);
+		return execute(new RedisCallback<List<V>>() {
+			@SuppressWarnings("unchecked")
+			@Override
+			public List<V> doInRedis(RedisConnection connection) {
+				return deserializeValues(connection.lRange(rawKey, start, end), List.class);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long remove(K key, final long count, Object value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.lRem(rawKey, count, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public V rightPop(K key) {
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.rPop(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public V rightPop(K key, long timeout, TimeUnit unit) {
+		final int tm = (int) unit.toSeconds(timeout);
+
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.bRPop(tm, rawKey).get(0);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long rightPush(K key, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.rPush(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long rightPushIfPresent(K key, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.rPushX(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long rightPush(K key, V pivot, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawPivot = rawValue(pivot);
+		final byte[] rawValue = rawValue(value);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.lInsert(rawKey, Position.AFTER, rawPivot, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public V rightPopAndLeftPush(K sourceKey, K destinationKey) {
+		final byte[] rawDestKey = rawKey(destinationKey);
+
+		return execute(new ValueDeserializingRedisCallback(sourceKey) {
+			@Override
+			protected byte[] inRedis(byte[] rawSourceKey, RedisConnection connection) {
+				return connection.rPopLPush(rawSourceKey, rawDestKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public V rightPopAndLeftPush(K sourceKey, K destinationKey, long timeout, TimeUnit unit) {
+		final int tm = (int) unit.toSeconds(timeout);
+		final byte[] rawDestKey = rawKey(destinationKey);
+
+		return execute(new ValueDeserializingRedisCallback(sourceKey) {
+			@Override
+			protected byte[] inRedis(byte[] rawSourceKey, RedisConnection connection) {
+				return connection.bRPopLPush(tm, rawSourceKey, rawDestKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public void set(K key, final long index, V value) {
+		final byte[] rawValue = rawValue(value);
+		execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				connection.lSet(rawKey, index, rawValue);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public void trim(K key, final long start, final long end) {
+		execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				connection.lTrim(rawKey, start, end);
+				return null;
+			}
+		}, true);
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultSetOperations.java

@@ -0,0 +1,241 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+
+/**
+ * Default implementation of {@link SetOperations}.
+ *  
+ * @author Costin Leau
+ */
+class DefaultSetOperations<K, V> extends AbstractOperations<K, V> implements SetOperations<K, V> {
+
+	public DefaultSetOperations(RedisTemplate<K, V> template) {
+		super(template);
+	}
+
+	@Override
+	public Boolean add(K key, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.sAdd(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Set<V> difference(K key, K otherKey) {
+		return difference(key, Collections.singleton(otherKey));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> difference(final K key, final Collection<K> otherKeys) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.sDiff(rawKeys);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@Override
+	public void differenceAndStore(K key, K otherKey, K destKey) {
+		differenceAndStore(key, Collections.singleton(otherKey), destKey);
+	}
+
+	@Override
+	public void differenceAndStore(final K key, final Collection<K> otherKeys, K destKey) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		final byte[] rawDestKey = rawKey(destKey);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.sDiffStore(rawDestKey, rawKeys);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public Set<V> intersect(K key, K otherKey) {
+		return intersect(key, Collections.singleton(otherKey));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> intersect(K key, Collection<K> otherKeys) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.sInter(rawKeys);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@Override
+	public void intersectAndStore(K key, K otherKey, K destKey) {
+		intersectAndStore(key, Collections.singleton(otherKey), destKey);
+	}
+
+	@Override
+	public void intersectAndStore(K key, Collection<K> otherKeys, K destKey) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		final byte[] rawDestKey = rawKey(destKey);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.sInterStore(rawDestKey, rawKeys);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public Boolean isMember(K key, Object o) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(o);
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.sIsMember(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> members(K key) {
+		final byte[] rawKey = rawKey(key);
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.sMembers(rawKey);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@Override
+	public Boolean move(K key, V value, K destKey) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawDestKey = rawKey(destKey);
+		final byte[] rawValue = rawValue(value);
+
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.sMove(rawKey, rawDestKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public V randomMember(K key) {
+
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.randomKey();
+			}
+		}, true);
+	}
+
+	@Override
+	public Boolean remove(K key, Object o) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(o);
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.sRem(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public V pop(K key) {
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.sPop(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long size(K key) {
+		final byte[] rawKey = rawKey(key);
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.sCard(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public Set<V> union(K key, K otherKey) {
+		return union(key, Collections.singleton(otherKey));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> union(K key, Collection<K> otherKeys) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.sUnion(rawKeys);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@Override
+	public void unionAndStore(K key, K otherKey, K destKey) {
+		unionAndStore(key, Collections.singleton(otherKey), destKey);
+	}
+
+	@Override
+	public void unionAndStore(K key, Collection<K> otherKeys, K destKey) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		final byte[] rawDestKey = rawKey(destKey);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.sUnionStore(rawDestKey, rawKeys);
+				return null;
+			}
+		}, true);
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultValueOperations.java

@@ -0,0 +1,243 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+
+/**
+ * Default implementation of {@link ValueOperations}.
+ * 
+ * @author Costin Leau
+ */
+class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements ValueOperations<K, V> {
+
+	DefaultValueOperations(RedisTemplate<K, V> template) {
+		super(template);
+	}
+
+	@Override
+	public V get(final Object key) {
+
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.get(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public V getAndSet(K key, V newValue) {
+		final byte[] rawValue = rawValue(newValue);
+		return execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.getSet(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long increment(K key, final long delta) {
+		final byte[] rawKey = rawKey(key);
+		// TODO add conversion service in here ?
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				if (delta == 1) {
+					return connection.incr(rawKey);
+				}
+
+				if (delta == -1) {
+					return connection.decr(rawKey);
+				}
+
+				if (delta < 0) {
+					return connection.decrBy(rawKey, delta);
+				}
+
+				return connection.incrBy(rawKey, delta);
+			}
+		}, true);
+	}
+
+	@Override
+	public Integer append(K key, String value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawString = rawString(value);
+
+		return execute(new RedisCallback<Integer>() {
+			@Override
+			public Integer doInRedis(RedisConnection connection) {
+				return connection.append(rawKey, rawString).intValue();
+			}
+		}, true);
+	}
+
+	@Override
+	public String get(K key, final int start, final int end) {
+		final byte[] rawKey = rawKey(key);
+
+		byte[] rawReturn = execute(new RedisCallback<byte[]>() {
+			@Override
+			public byte[] doInRedis(RedisConnection connection) {
+				return connection.getRange(rawKey, start, end);
+			}
+		}, true);
+
+		return deserializeString(rawReturn);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Collection<V> multiGet(Collection<K> keys) {
+		if (keys.isEmpty()) {
+			return Collections.emptyList();
+		}
+
+		final byte[][] rawKeys = new byte[keys.size()][];
+
+		int counter = 0;
+		for (K hashKey : keys) {
+			rawKeys[counter++] = rawKey(hashKey);
+		}
+
+		List<byte[]> rawValues = execute(new RedisCallback<List<byte[]>>() {
+			@Override
+			public List<byte[]> doInRedis(RedisConnection connection) {
+				return connection.mGet(rawKeys);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, List.class);
+	}
+
+	@Override
+	public void multiSet(Map<? extends K, ? extends V> m) {
+		if (m.isEmpty()) {
+			return;
+		}
+
+		final Map<byte[], byte[]> rawKeys = new LinkedHashMap<byte[], byte[]>(m.size());
+
+		for (Map.Entry<? extends K, ? extends V> entry : m.entrySet()) {
+			rawKeys.put(rawKey(entry.getKey()), rawValue(entry.getValue()));
+		}
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.mSet(rawKeys);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public void multiSetIfAbsent(Map<? extends K, ? extends V> m) {
+		if (m.isEmpty()) {
+			return;
+		}
+
+		final Map<byte[], byte[]> rawKeys = new LinkedHashMap<byte[], byte[]>(m.size());
+
+		for (Map.Entry<? extends K, ? extends V> entry : m.entrySet()) {
+			rawKeys.put(rawKey(entry.getKey()), rawValue(entry.getValue()));
+		}
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.mSetNX(rawKeys);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public void set(K key, V value) {
+		final byte[] rawValue = rawValue(value);
+		execute(new ValueDeserializingRedisCallback(key) {
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				connection.set(rawKey, rawValue);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public void set(K key, V value, long timeout, TimeUnit unit) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+		final long rawTimeout = unit.toSeconds(timeout);
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) throws DataAccessException {
+				connection.setEx(rawKey, (int) rawTimeout, rawValue);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public Boolean setIfAbsent(K key, V value) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
+				return connection.setNX(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+
+	@Override
+	public void set(K key, final int start, final int end) {
+		final byte[] rawKey = rawKey(key);
+
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.setRange(rawKey, start, end);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public Long size(K key) {
+		final byte[] rawKey = rawKey(key);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.strLen(rawKey);
+			}
+		}, true);
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultZSetOperations.java

@@ -0,0 +1,243 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+
+/**
+ * Default implementation of {@link ZSetOperations}.
+ * 
+ * @author Costin Leau
+ */
+class DefaultZSetOperations<K, V> extends AbstractOperations<K, V> implements ZSetOperations<K, V> {
+
+	DefaultZSetOperations(RedisTemplate<K, V> template) {
+		super(template);
+	}
+
+	@Override
+	public Boolean add(final K key, final V value, final double score) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.zAdd(rawKey, score, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Double incrementScore(K key, V value, final double delta) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(value);
+
+		return execute(new RedisCallback<Double>() {
+			@Override
+			public Double doInRedis(RedisConnection connection) {
+				return connection.zIncrBy(rawKey, delta, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public void intersectAndStore(K key, K otherKey, K destKey) {
+		intersectAndStore(key, Collections.singleton(otherKey), destKey);
+	}
+
+	@Override
+	public void intersectAndStore(K key, Collection<K> otherKeys, K destKey) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		final byte[] rawDestKey = rawKey(destKey);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.zInterStore(rawDestKey, rawKeys);
+				return null;
+			}
+		}, true);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> range(K key, final long start, final long end) {
+		final byte[] rawKey = rawKey(key);
+
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.zRange(rawKey, start, end);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> rangeByScore(K key, final double min, final double max) {
+		final byte[] rawKey = rawKey(key);
+
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.zRangeByScore(rawKey, min, max);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@Override
+	public Long rank(K key, Object o) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(o);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				Long zRank = connection.zRank(rawKey, rawValue);
+				return (zRank != null && zRank.longValue() >= 0 ? zRank : null);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long reverseRank(K key, Object o) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(o);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				Long zRank = connection.zRevRank(rawKey, rawValue);
+				return (zRank != null && zRank.longValue() >= 0 ? zRank : null);
+			}
+		}, true);
+	}
+
+	@Override
+	public Boolean remove(K key, Object o) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(o);
+
+		return execute(new RedisCallback<Boolean>() {
+			@Override
+			public Boolean doInRedis(RedisConnection connection) {
+				return connection.zRem(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public void removeRange(K key, final long start, final long end) {
+		final byte[] rawKey = rawKey(key);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.zRemRange(rawKey, start, end);
+				return null;
+			}
+		}, true);
+	}
+
+	@Override
+	public void removeRangeByScore(K key, final double min, final double max) {
+		final byte[] rawKey = rawKey(key);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.zRemRangeByScore(rawKey, min, max);
+				return null;
+			}
+		}, true);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Set<V> reverseRange(K key, final long start, final long end) {
+		final byte[] rawKey = rawKey(key);
+
+		Set<byte[]> rawValues = execute(new RedisCallback<Set<byte[]>>() {
+			@Override
+			public Set<byte[]> doInRedis(RedisConnection connection) {
+				return connection.zRevRange(rawKey, start, end);
+			}
+		}, true);
+
+		return deserializeValues(rawValues, Set.class);
+	}
+
+	@Override
+	public Double score(K key, Object o) {
+		final byte[] rawKey = rawKey(key);
+		final byte[] rawValue = rawValue(o);
+
+		return execute(new RedisCallback<Double>() {
+			@Override
+			public Double doInRedis(RedisConnection connection) {
+				return connection.zScore(rawKey, rawValue);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long count(K key, final double min, final double max) {
+		final byte[] rawKey = rawKey(key);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.zCount(rawKey, min, max);
+			}
+		}, true);
+	}
+
+	@Override
+	public Long size(K key) {
+		final byte[] rawKey = rawKey(key);
+
+		return execute(new RedisCallback<Long>() {
+			@Override
+			public Long doInRedis(RedisConnection connection) {
+				return connection.zCard(rawKey);
+			}
+		}, true);
+	}
+
+	@Override
+	public void unionAndStore(K key, K otherKey, K destKey) {
+		unionAndStore(key, Collections.singleton(otherKey), destKey);
+	}
+
+	@Override
+	public void unionAndStore(K key, Collection<K> otherKeys, K destKey) {
+		final byte[][] rawKeys = rawKeys(key, otherKeys);
+		final byte[] rawDestKey = rawKey(destKey);
+		execute(new RedisCallback<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.zUnionStore(rawDestKey, rawKeys);
+				return null;
+			}
+		}, true);
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/SerializationUtils.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+
+import org.springframework.data.keyvalue.redis.connection.DefaultSortParameters;
+import org.springframework.data.keyvalue.redis.connection.SortParameters;
+import org.springframework.data.keyvalue.redis.core.query.SortQuery;
+import org.springframework.data.keyvalue.redis.serializer.RedisSerializer;
+
+/**
+ * Utility class with various serialization-related methods. 
+ * 
+ * @author Costin Leau
+ */
+public abstract class SerializationUtils {
+
+	public static <T> T deserialize(byte[] value, RedisSerializer<T> serializer) {
+		if (isEmpty(value)) {
+			return null;
+		}
+		return serializer.deserialize(value);
+	}
+
+	@SuppressWarnings("unchecked")
+	static <T extends Collection<?>> T deserializeValues(Collection<byte[]> rawValues, Class<T> type, RedisSerializer<?> redisSerializer) {
+		Collection<Object> values = (List.class.isAssignableFrom(type) ? new ArrayList<Object>(rawValues.size())
+				: new LinkedHashSet<Object>(rawValues.size()));
+		for (byte[] bs : rawValues) {
+			if (bs != null) {
+				values.add(redisSerializer.deserialize(bs));
+			}
+		}
+
+		return (T) values;
+	}
+
+	public static boolean isEmpty(byte[] data) {
+		return (data == null || data.length == 0);
+	}
+
+	public static <K> SortParameters convertQuery(SortQuery<K> query, RedisSerializer<String> stringSerializer) {
+
+		return new DefaultSortParameters(stringSerializer.serialize(query.getBy()), query.getLimit(), serialize(
+				query.getGetPattern(), stringSerializer), query.getOrder(), query.isAlphabetic());
+	}
+
+	public static byte[][] serialize(List<String> strings, RedisSerializer<String> stringSerializer) {
+		List<byte[]> raw = null;
+
+		if (strings == null) {
+			raw = Collections.emptyList();
+		}
+		else {
+			raw = new ArrayList<byte[]>(strings.size());
+			for (String key : strings) {
+				raw.add(stringSerializer.serialize(key));
+			}
+		}
+		return raw.toArray(new byte[raw.size()][]);
+	}
+}