Add support for LPOP and RPOP with count option.

Closes #1987

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

@@ -697,6 +697,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.lPop(key), Converters.identityConverter());
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#lPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> lPop(byte[] key, long count) {
+		return convertAndReturn(delegate.lPop(key, count), Converters.identityConverter());
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#lPos(byte[], byte[], java.lang.Integer, java.lang.Integer)
@@ -896,6 +905,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.rPop(key), Converters.identityConverter());
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> rPop(byte[] key, long count) {
+		return convertAndReturn(delegate.rPop(key, count), Converters.identityConverter());
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#rPopLPush(byte[], byte[])
@@ -2169,6 +2187,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.lPop(serialize(key)), bytesToString);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#lPop(java.lang.String, long)
+	 */
+	@Override
+	public List<String> lPop(String key, long count) {
+		return convertAndReturn(delegate.lPop(serialize(key), count), byteListToStringList);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#lPos(java.lang.String, java.lang.String, java.lang.Integer, java.lang.Integer)
@@ -2322,6 +2349,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.rPop(serialize(key)), bytesToString);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#rPop(java.lang.String, long)
+	 */
+	@Override
+	public List<String> rPop(String key, long count) {
+		return convertAndReturn(delegate.rPop(serialize(key), count), byteListToStringList);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#rPopLPush(java.lang.String, java.lang.String)

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

@@ -56,6 +56,7 @@ import org.springframework.lang.Nullable;
  * @author Mark Paluch
  * @author Tugdual Grall
  * @author Andrey Shlykov
+ * @author dengliming
  * @since 2.0
  */
 public interface DefaultedRedisConnection extends RedisConnection {
@@ -712,6 +713,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return listCommands().lPop(key);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#listCommands()}}. */
+	@Override
+	@Deprecated
+	default List<byte[]> lPop(byte[] key, long count) {
+		return listCommands().lPop(key, count);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#listCommands()}}. */
 	@Override
 	@Deprecated
@@ -719,6 +727,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return listCommands().rPop(key);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#listCommands()}}. */
+	@Override
+	@Deprecated
+	default List<byte[]> rPop(byte[] key, long count) {
+		return listCommands().rPop(key, count);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#listCommands()}}. */
 	@Override
 	@Deprecated

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

@@ -42,6 +42,7 @@ import org.springframework.util.ObjectUtils;
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  * @since 2.0
  */
 public interface ReactiveListCommands {
@@ -874,12 +875,14 @@ public interface ReactiveListCommands {
 	 */
 	class PopCommand extends KeyCommand {
 
+		private final long count;
+
 		private final Direction direction;
 
-		private PopCommand(@Nullable ByteBuffer key, Direction direction) {
+		private PopCommand(@Nullable ByteBuffer key, long count, Direction direction) {
 
 			super(key);
-
+			this.count = count;
 			this.direction = direction;
 		}
 
@@ -889,7 +892,7 @@ public interface ReactiveListCommands {
 		 * @return a new {@link PopCommand} for right push ({@literal RPOP}).
 		 */
 		public static PopCommand right() {
-			return new PopCommand(null, Direction.RIGHT);
+			return new PopCommand(null, 0, Direction.RIGHT);
 		}
 
 		/**
@@ -898,7 +901,7 @@ public interface ReactiveListCommands {
 		 * @return a new {@link PopCommand} for right push ({@literal LPOP}).
 		 */
 		public static PopCommand left() {
-			return new PopCommand(null, Direction.LEFT);
+			return new PopCommand(null, 0, Direction.LEFT);
 		}
 
 		/**
@@ -911,7 +914,17 @@ public interface ReactiveListCommands {
 
 			Assert.notNull(key, "Key must not be null!");
 
-			return new PopCommand(key, direction);
+			return new PopCommand(key, count, direction);
+		}
+
+		/**
+		 * Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
+		 *
+		 * @param count
+		 * @return a new {@link LSetCommand} with {@literal value} applied.
+		 */
+		public PopCommand count(long count) {
+			return new PopCommand(getKey(), count, direction);
 		}
 
 		/**
@@ -920,6 +933,10 @@ public interface ReactiveListCommands {
 		public Direction getDirection() {
 			return direction;
 		}
+
+		public long getCount() {
+			return count;
+		}
 	}
 
 	/**
@@ -936,6 +953,21 @@ public interface ReactiveListCommands {
 		return pop(Mono.just(PopCommand.left().from(key))).next().map(ByteBufferResponse::getOutput);
 	}
 
+	/**
+	 * Removes and returns first element in list stored at {@literal key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return
+	 * @see <a href="https://redis.io/commands/lpop">Redis Documentation: LPOP</a>
+	 */
+	default Flux<ByteBuffer> lPop(ByteBuffer key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return popList(Mono.just(PopCommand.left().from(key).count(count))).flatMap(CommandResponse::getOutput);
+	}
+
 	/**
 	 * Removes and returns last element in list stored at {@literal key}.
 	 *
@@ -950,6 +982,21 @@ public interface ReactiveListCommands {
 		return pop(Mono.just(PopCommand.right().from(key))).next().map(ByteBufferResponse::getOutput);
 	}
 
+	/**
+	 * Removes and returns last element in list stored at {@literal key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return
+	 * @see <a href="https://redis.io/commands/rpop">Redis Documentation: RPOP</a>
+	 */
+	default Flux<ByteBuffer> rPop(ByteBuffer key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return popList(Mono.just(PopCommand.right().from(key).count(count))).flatMap(CommandResponse::getOutput);
+	}
+
 	/**
 	 * Removes and returns last element in list stored at {@link KeyCommand#getKey()}
 	 *
@@ -960,6 +1007,16 @@ public interface ReactiveListCommands {
 	 */
 	Flux<ByteBufferResponse<PopCommand>> pop(Publisher<PopCommand> commands);
 
+	/**
+	 * Removes and returns last element in list stored at {@link KeyCommand#getKey()}
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/lpop">Redis Documentation: LPOP</a>
+	 * @see <a href="https://redis.io/commands/rpop">Redis Documentation: RPOP</a>
+	 */
+	Flux<CommandResponse<PopCommand, Flux<ByteBuffer>>> popList(Publisher<PopCommand> commands);
+
 	/**
 	 * @author Christoph Strobl
 	 * @see <a href="https://redis.io/commands/blpop">Redis Documentation: BLPOP</a>

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

@@ -26,6 +26,7 @@ import org.springframework.util.CollectionUtils;
  * @author Costin Leau
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  */
 public interface RedisListCommands {
 
@@ -200,6 +201,17 @@ public interface RedisListCommands {
 	@Nullable
 	byte[] lPop(byte[] key);
 
+	/**
+	 * Removes and returns first element in list stored at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return {@literal null} when key does not exist or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/lpop">Redis Documentation: LPOP</a>
+	 */
+	@Nullable
+	List<byte[]> lPop(byte[] key, long count);
+
 	/**
 	 * Removes and returns last element in list stored at {@code key}.
 	 *
@@ -210,6 +222,17 @@ public interface RedisListCommands {
 	@Nullable
 	byte[] rPop(byte[] key);
 
+	/**
+	 * Removes and returns last element in list stored at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return {@literal null} when key does not exist or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/rpop">Redis Documentation: RPOP</a>
+	 */
+	@Nullable
+	List<byte[]> rPop(byte[] key, long count);
+
 	/**
 	 * Removes and returns first element from lists stored at {@code keys}. <br>
 	 * <b>Blocks connection</b> until element available or {@code timeout} reached.

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

@@ -842,6 +842,17 @@ public interface StringRedisConnection extends RedisConnection {
 	 */
 	String lPop(String key);
 
+	/**
+	 * Removes and returns first element in list stored at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return
+	 * @see <a href="https://redis.io/commands/lpop">Redis Documentation: LPOP</a>
+	 * @see RedisListCommands#lPop(byte[], long)
+	 */
+	List<String> lPop(String key, long count);
+
 	/**
 	 * Removes and returns last element in list stored at {@code key}.
 	 *
@@ -852,6 +863,17 @@ public interface StringRedisConnection extends RedisConnection {
 	 */
 	String rPop(String key);
 
+	/**
+	 * Removes and returns last element in list stored at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return
+	 * @see <a href="https://redis.io/commands/rpop">Redis Documentation: RPOP</a>
+	 * @see RedisListCommands#rPop(byte[], long)
+	 */
+	List<String> rPop(String key, long count);
+
 	/**
 	 * Removes and returns first element from lists stored at {@code keys} (see: {@link #lPop(byte[])}). <br>
 	 * <b>Blocks connection</b> until element available or {@code timeout} reached.

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

@@ -33,6 +33,7 @@ import org.springframework.util.CollectionUtils;
  * @author Christoph Strobl
  * @author Mark Paluch
  * @author Jot Zhao
+ * @author dengliming
  * @since 2.0
  */
 class JedisClusterListCommands implements RedisListCommands {
@@ -269,6 +270,22 @@ class JedisClusterListCommands implements RedisListCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#lPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> lPop(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			return connection.getCluster().lpop(key, (int) count);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[])
@@ -285,6 +302,22 @@ class JedisClusterListCommands implements RedisListCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> rPop(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			return connection.getCluster().rpop(key, (int) count);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#bLPop(int, byte[][])

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

@@ -30,6 +30,7 @@ import org.springframework.util.Assert;
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  * @since 2.0
  */
 class JedisListCommands implements RedisListCommands {
@@ -214,6 +215,18 @@ class JedisListCommands implements RedisListCommands {
 		return connection.invoke().just(BinaryJedis::lpop, MultiKeyPipelineBase::lpop, key);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#lPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> lPop(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().just(BinaryJedis::lpop, MultiKeyPipelineBase::lpop, key, (int) count);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[])
@@ -226,6 +239,18 @@ class JedisListCommands implements RedisListCommands {
 		return connection.invoke().just(BinaryJedis::rpop, MultiKeyPipelineBase::rpop, key);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> rPop(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().just(BinaryJedis::rpop, MultiKeyPipelineBase::rpop, key, (int) count);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#bLPop(int, byte[][])

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

@@ -30,6 +30,7 @@ import org.springframework.util.Assert;
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  * @since 2.0
  */
 class LettuceListCommands implements RedisListCommands {
@@ -213,6 +214,18 @@ class LettuceListCommands implements RedisListCommands {
 		return connection.invoke().just(RedisListAsyncCommands::lpop, key);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#lPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> lPop(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().just(RedisListAsyncCommands::lpop, key, count);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[])
@@ -225,6 +238,18 @@ class LettuceListCommands implements RedisListCommands {
 		return connection.invoke().just(RedisListAsyncCommands::rpop, key);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisListCommands#rPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> rPop(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().just(RedisListAsyncCommands::rpop, key, count);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisListCommands#bLPop(int, byte[][])

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

@@ -41,6 +41,7 @@ import org.springframework.util.ObjectUtils;
  * @author Christoph Strobl
  * @author Mark Paluch
  * @author Michele Mancioppi
+ * @author dengliming
  * @since 2.0
  */
 class LettuceReactiveListCommands implements ReactiveListCommands {
@@ -252,7 +253,7 @@ class LettuceReactiveListCommands implements ReactiveListCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveListCommands#rPop(org.reactivestreams.Publisher)
+	 * @see org.springframework.data.redis.connection.ReactiveListCommands#pop(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<ByteBufferResponse<PopCommand>> pop(Publisher<PopCommand> commands) {
@@ -270,6 +271,25 @@ class LettuceReactiveListCommands implements ReactiveListCommands {
 		}));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveListCommands#popList(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<PopCommand, Flux<ByteBuffer>>> popList(Publisher<PopCommand> commands) {
+		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+			Assert.notNull(command.getDirection(), "Direction must not be null!");
+
+			Flux<ByteBuffer> popResult = ObjectUtils.nullSafeEquals(Direction.RIGHT, command.getDirection())
+					? cmd.rpop(command.getKey(), command.getCount())
+					: cmd.lpop(command.getKey(), command.getCount());
+
+			return Mono.just(new CommandResponse<>(command, popResult));
+		}));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveListCommands#bPop(org.reactivestreams.Publisher)

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

@@ -30,6 +30,7 @@ import org.springframework.util.CollectionUtils;
  * @author David Liu
  * @author Thomas Darimont
  * @author Christoph Strobl
+ * @author dengliming
  */
 class DefaultListOperations<K, V> extends AbstractOperations<K, V> implements ListOperations<K, V> {
 
@@ -97,6 +98,16 @@ class DefaultListOperations<K, V> extends AbstractOperations<K, V> implements Li
 		}, true);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ListOperations#leftPop(java.lang.Object, long)
+	 */
+	@Override
+	public List<V> leftPop(K key, long count) {
+		byte[] rawKey = rawKey(key);
+		return execute(connection -> deserializeValues(connection.lPop(rawKey, count)), true);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ListOperations#leftPop(java.lang.Object, long, java.util.concurrent.TimeUnit)
@@ -227,6 +238,16 @@ class DefaultListOperations<K, V> extends AbstractOperations<K, V> implements Li
 		}, true);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ListOperations#rightPop(java.lang.Object, long)
+	 */
+	@Override
+	public List<V> rightPop(K key, long count) {
+		byte[] rawKey = rawKey(key);
+		return execute(connection -> deserializeValues(connection.rPop(rawKey, count)), true);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ListOperations#rightPop(java.lang.Object, long, java.util.concurrent.TimeUnit)

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

@@ -31,6 +31,7 @@ import org.springframework.util.Assert;
  * @author Thomas Darimont
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  */
 public interface ListOperations<K, V> {
 
@@ -247,6 +248,17 @@ public interface ListOperations<K, V> {
 	@Nullable
 	V leftPop(K key);
 
+	/**
+	 * Removes and returns first element in list stored at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/lpop">Redis Documentation: LPOP</a>
+	 */
+	@Nullable
+	List<V> leftPop(K key, long count);
+
 	/**
 	 * Removes and returns first element from lists stored at {@code key} . <br>
 	 * <b>Blocks connection</b> until element available or {@code timeout} reached.
@@ -290,6 +302,17 @@ public interface ListOperations<K, V> {
 	@Nullable
 	V rightPop(K key);
 
+	/**
+	 * Removes and returns last element in list stored at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/rpop">Redis Documentation: RPOP</a>
+	 */
+	@Nullable
+	List<V> rightPop(K key, long count);
+
 	/**
 	 * Removes and returns last element from lists stored at {@code key}. <br>
 	 * <b>Blocks connection</b> until element available or {@code timeout} reached.