DATAKV-27

+ add initial pipeline support to RedisConnection and Jedis and JRedis implementations

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisConnection.java

@@ -16,6 +16,8 @@
 
 package org.springframework.data.keyvalue.redis.connection;
 
+import java.util.List;
+
 import org.springframework.data.keyvalue.redis.UncategorizedRedisException;
 
 /**
@@ -54,10 +56,44 @@ public interface RedisConnection extends RedisCommands {
 	 * Indicates whether the connection is in "queue"(or "MULTI") mode or not.
 	 * When queueing, all commands are postponed until EXEC or DISCARD commands
 	 * are issued.
-	 * Since in queueing, no results are returned, the connection will return NULL
+	 * Since in queueing no results are returned, the connection will return NULL
 	 * on all operations that interact with the data. 
 	 * 
 	 * @return true if the connection is in queue/MULTI mode, false otherwise
 	 */
 	boolean isQueueing();
-}
+
+	/**
+	 * Indicates whether the connection is currently pipelined or not.
+	 *  
+	 * @return true if the connection is pipelined, false otherwise
+	 * @see #openPipeline()
+	 * @see #isQueueing()
+	 */
+	boolean isPipelined();
+
+	/**
+	 * Activates the pipeline mode for this connection. When pipelined, all commands return null
+	 * (the reply is read at the end through {@link #closePipeline()}.
+	 * Calling this method when the connection is already pipelined has no effect.
+	 * 
+	 * Pipelining is used for issuing commands without requesting the response right away but rather
+	 * at the end of the batch. While somewhat similar to MULTI, pipelining does not
+	 * guarantee atomicity - it only tries to improve performance when issuing a lot of
+	 * commands (such as in batching scenarios).
+	 * 
+	 * <p>Note:</p>Consider doing some performance testing before using this feature since
+	 * in many cases the performance benefits are minimal yet the impact on usage are not.
+	 * 
+	 * @see #multi()
+	 */
+	void openPipeline();
+
+	/**
+	 * Executes the commands in the pipeline and returns their result.
+	 * If the connection is not pipelined, an empty collection is returned.
+	 * 
+	 * @return the result of the executed commands.
+	 */
+	List<Object> closePipeline();
+}

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

@@ -19,6 +19,7 @@ import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -41,6 +42,7 @@ import redis.clients.jedis.BinaryTransaction;
 import redis.clients.jedis.Client;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisException;
+import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.Transaction;
 import redis.clients.jedis.ZParams;
@@ -67,6 +69,7 @@ public class JedisConnection implements RedisConnection {
 
 
 	private volatile JedisSubscription subscription;
+	private volatile Pipeline pipeline;
 
 	/**
 	 * Constructs a new <code>JedisConnection</code> instance.
@@ -152,6 +155,26 @@ public class JedisConnection implements RedisConnection {
 		return client.isInMulti();
 	}
 
+	@Override
+	public boolean isPipelined() {
+		return (pipeline != null);
+	}
+
+	@Override
+	public void openPipeline() {
+		if (pipeline == null) {
+			pipeline = jedis.pipelined();
+		}
+	}
+
+	@Override
+	public List<Object> closePipeline() {
+		if (pipeline != null) {
+			return pipeline.execute();
+		}
+		return Collections.emptyList();
+	}
+
 	@Override
 	public List<byte[]> sort(byte[] key, SortParameters params) {
 

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

@@ -17,6 +17,7 @@ package org.springframework.data.keyvalue.redis.connection.jredis;
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
@@ -47,6 +48,7 @@ public class JredisConnection implements RedisConnection {
 
 	private final JRedis jredis;
 	private final boolean isPool;
+	private boolean isClosed = false;
 
 	/**
 	 * Constructs a new <code>JredisConnection</code> instance.
@@ -69,6 +71,8 @@ public class JredisConnection implements RedisConnection {
 
 	@Override
 	public void close() throws UncategorizedRedisException {
+		isClosed = true;
+
 		// don't actually close the connection
 		// if a pool is used
 		if (!isPool) {
@@ -83,7 +87,7 @@ public class JredisConnection implements RedisConnection {
 
 	@Override
 	public boolean isClosed() {
-		throw new UnsupportedOperationException();
+		return isClosed;
 	}
 
 	@Override
@@ -91,6 +95,21 @@ public class JredisConnection implements RedisConnection {
 		return false;
 	}
 
+	@Override
+	public boolean isPipelined() {
+		return false;
+	}
+
+	@Override
+	public void openPipeline() {
+		throw new UnsupportedOperationException("Pipelining not supported by JRedis");
+	}
+
+	@Override
+	public List<Object> closePipeline() {
+		return Collections.emptyList();
+	}
+
 	@Override
 	public List<byte[]> sort(byte[] key, SortParameters params) {
 		Sort sort = jredis.sort(JredisUtils.decode(key));