DATAKV-43

+ introduce support for RW pipelined callbacks (in addition to the WO support)

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

@@ -64,6 +64,17 @@ public interface RedisOperations<K, V> {
 	 */
 	<T> T execute(SessionCallback<T> session);
 
+	/**
+	 * Executes the given action object on a pipelined connection, returning the results. Note that the callback <b>cannot</b>
+	 * return a non-null value as it gets overwritten by the pipeline.
+	 * 
+	 * @param <T> list element return type
+	 * @param action callback object to execute 
+	 * @return list of objects returned by the pipeline
+	 */
+	List<V> executePipelined(RedisCallback<?> action);
+
+
 	Boolean hasKey(K key);
 
 	void delete(K key);

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

@@ -25,6 +25,7 @@ import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.keyvalue.redis.connection.DataType;
 import org.springframework.data.keyvalue.redis.connection.RedisConnection;
 import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
@@ -151,12 +152,14 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 	}
 
 	/**
-	 * Executes the given action object within a connection, that can be pipelined or not and which can be exposed or not.
+	 * Executes the given action object within a connection that can be exposed or not. Additionally, the connection
+	 * can be pipelined. Note the results of the pipeline are discarded (making it suitable for write-only scenarios).
+	 * Use {@link #executePipelined(RedisCallback)} as an alternative.
 	 * 
 	 * @param <T> return type
 	 * @param action callback object to execute
 	 * @param exposeConnection whether to enforce exposure of the native Redis Connection to callback code
-	 * @param pipeline whether to pipeline or not the connection for the execution duration 
+	 * @param pipeline whether to pipeline or not the connection for the execution 
 	 * @return object returned by the action
 	 */
 	public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline) {
@@ -189,6 +192,48 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		}
 	}
 
+
+	@Override
+	public <T> T execute(SessionCallback<T> session) {
+		RedisConnectionFactory factory = getConnectionFactory();
+		// bind connection
+		RedisConnectionUtils.bindConnection(factory);
+		try {
+			return session.execute(this);
+		} finally {
+			RedisConnectionUtils.unbindConnection(factory);
+		}
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public List<V> executePipelined(final RedisCallback<?> action) {
+		return executePipelined(action, valueSerializer);
+	}
+
+	/**
+	 * Executes the given action object on a pipelined connection, returning the results using a dedicated serializer.
+	 * Note that the callback <b>cannot</b> return a non-null value as it gets overwritten by the pipeline.
+	 * 
+	 * @param action callback object to execute
+	 * @param resultSerializer
+	 * @return list of objects returned by the pipeline
+	 */
+	public <T> List<T> executePipelined(final RedisCallback<?> action, final RedisSerializer<T> resultSerializer) {
+		return execute(new RedisCallback<List<T>>() {
+			public List<T> doInRedis(RedisConnection connection) throws DataAccessException {
+				connection.openPipeline();
+				Object result = action.doInRedis(connection);
+				if (result != null) {
+					throw new InvalidDataAccessApiUsageException(
+							"Callback cannot returned a non-null value as it gets overwritten by the pipeline");
+				}
+				List<byte[]> pipeline = connection.closePipeline();
+				return SerializationUtils.deserialize(pipeline, resultSerializer);
+			}
+		});
+	}
+
 	protected RedisConnection createRedisConnectionProxy(RedisConnection pm) {
 		Class<?>[] ifcs = ClassUtils.getAllInterfacesForClass(pm.getClass(), getClass().getClassLoader());
 		return (RedisConnection) Proxy.newProxyInstance(pm.getClass().getClassLoader(), ifcs,
@@ -208,18 +253,6 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return result;
 	}
 
-	@Override
-	public <T> T execute(SessionCallback<T> session) {
-		RedisConnectionFactory factory = getConnectionFactory();
-		// bind connection
-		RedisConnectionUtils.bindConnection(factory);
-		try {
-			return session.execute(this);
-		} finally {
-			RedisConnectionUtils.unbindConnection(factory);
-		}
-	}
-
 	/**
 	 * Returns whether to expose the native Redis connection to RedisCallback code, or rather a connection proxy (the default). 
 	 *

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.keyvalue.redis.core;
 
+import org.springframework.dao.DataAccessException;
+
 /**
  * Callback executing all operations against a surrogate 'session' (basically against the same underlying Redis connection).
  * Allows 'transactions' to take place through the use of multi/discard/exec/watch/unwatch commands. 
@@ -29,5 +31,5 @@ public interface SessionCallback<T> {
 	 * @param operations Redis operations
 	 * @return return value
 	 */
-	<K, V> T execute(RedisOperations<K, V> operations);
+	<K, V> T execute(RedisOperations<K, V> operations) throws DataAccessException;
 }

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/core/SessionTest.java

@@ -36,7 +36,7 @@ public class SessionTest {
 		when(factory.getConnection()).thenReturn(conn);
 		final StringRedisTemplate template = new StringRedisTemplate(factory);
 
-		template.execute(new SessionCallback() {
+		template.execute(new SessionCallback<Object>() {
 			@Override
 			public Object execute(RedisOperations operations) {
 				checkConnection(template, conn);
@@ -48,7 +48,7 @@ public class SessionTest {
 		});
 	}
 
-	private void checkConnection(RedisTemplate template, final RedisConnection expectedConnection) {
+	private void checkConnection(RedisTemplate<?, ?> template, final RedisConnection expectedConnection) {
 		template.execute(new RedisCallback<Object>() {
 
 			@Override