DATAREDIS-668 - Add SPOP with COUNT command.

We now support SPOP command with COUNT option throughout the imperative, cluster and reactive API.

Original pull request: #259.

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

@@ -868,6 +868,20 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return result;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSetCommands#sPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> sPop(byte[] key, long count) {
+
+		List<byte[]> result = delegate.sPop(key, count);
+		if (isFutureConversion()) {
+			addResultConverter(identityConverter);
+		}
+		return result;
+	}
+
 	public byte[] sRandMember(byte[] key) {
 		byte[] result = delegate.sRandMember(key);
 		if (isFutureConversion()) {
@@ -2032,6 +2046,22 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return bytesToString.convert(result);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#sPop(java.lang.String, long)
+	 */
+	@Override
+	public List<String> sPop(String key, long count) {
+
+		List<byte[]> result = delegate.sPop(serialize(key), count);
+
+		if (isFutureConversion()) {
+			addResultConverter(byteListToStringList);
+		}
+
+		return byteListToStringList.convert(result);
+	}
+
 	public String sRandMember(String key) {
 		byte[] result = delegate.sRandMember(serialize(key));
 		if (isFutureConversion()) {

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

@@ -563,6 +563,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return setCommands().sPop(key);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#setCommands()}}. */
+	@Override
+	@Deprecated
+	default List<byte[]> sPop(byte[] key, long count) {
+		return setCommands().sPop(key, count);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#setCommands()}}. */
 	@Override
 	@Deprecated

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

@@ -251,6 +251,39 @@ public interface ReactiveSetCommands {
 	 */
 	Flux<NumericResponse<SRemCommand, Long>> sRem(Publisher<SRemCommand> commands);
 
+	/**
+	 * {@code SPOP} command parameters.
+	 *
+	 * @author Christoph Strobl
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 */
+	class SPopCommand extends KeyCommand {
+
+		private final long count;
+
+		private SPopCommand(ByteBuffer key, long count) {
+
+			super(key);
+			this.count = count;
+		}
+
+		public static SPopCommand one() {
+			return new SPopCommand(null, 1L);
+		}
+
+		public static SPopCommand members(long count) {
+			return new SPopCommand(null, count);
+		}
+
+		public SPopCommand from(ByteBuffer key) {
+			return new SPopCommand(key, count);
+		}
+
+		public long getCount() {
+			return count;
+		}
+	}
+
 	/**
 	 * Remove and return a random member from set at {@literal key}.
 	 *
@@ -262,9 +295,32 @@ public interface ReactiveSetCommands {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return sPop(Mono.just(new KeyCommand(key))).next().map(ByteBufferResponse::getOutput);
+		return sPop(Mono.just(SPopCommand.one().from(key))).next().map(ByteBufferResponse::getOutput);
 	}
 
+	/**
+	 * Remove and return a random member from set at {@literal key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 */
+	default Flux<ByteBuffer> sPop(ByteBuffer key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return sPop(SPopCommand.members(count).from(key));
+	}
+
+	/**
+	 * Remove and return a random member from set at {@literal key}.
+	 *
+	 * @param command must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 */
+	Flux<ByteBuffer> sPop(SPopCommand command);
+
 	/**
 	 * Remove and return a random member from set at {@link KeyCommand#getKey()}
 	 *

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

@@ -59,6 +59,17 @@ public interface RedisSetCommands {
 	 */
 	byte[] sPop(byte[] key);
 
+	/**
+	 * Remove and return {@code count} random members from set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count the number of random members to pop from the set.
+	 * @return empty {@link List} if set does not exist. Never {@literal null}.
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 * @since 2.0
+	 */
+	List<byte[]> sPop(byte[] key, long count);
+
 	/**
 	 * Move {@code value} from {@code srcKey} to {@code destKey}
 	 *

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

@@ -781,6 +781,18 @@ public interface StringRedisConnection extends RedisConnection {
 	 */
 	String sPop(String key);
 
+	/**
+	 * Remove and return {@code count} random members from set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count the number of random members to return.
+	 * @return empty {@link List} if {@literal key} does not exist.
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 * @see RedisSetCommands#sPop(byte[], long)
+	 * @since 2.0
+	 */
+	List<String> sPop(String key, long count);
+
 	/**
 	 * Move {@code value} from {@code srcKey} to {@code destKey}
 	 *

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

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.jedis;
 
 import redis.clients.jedis.ScanParams;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -87,6 +88,20 @@ class JedisClusterSetCommands implements RedisSetCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSetCommands#sPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> sPop(byte[] key, long count) {
+
+		try {
+			return new ArrayList<>(connection.getCluster().spop(key, count));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisSetCommands#sMove(byte[], byte[], byte[])

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

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.jedis;
 
 import redis.clients.jedis.ScanParams;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
@@ -251,6 +252,28 @@ class JedisSetCommands implements RedisSetCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSetCommands#sPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> sPop(byte[] key, long count) {
+
+		try {
+			if (isPipelined()) {
+				pipeline(connection.newJedisResult(connection.getPipeline().spop(key, count), ArrayList::new));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection.newJedisResult(connection.getTransaction().spop(key, count), ArrayList::new));
+				return null;
+			}
+			return new ArrayList<>(connection.getJedis().spop(key, count));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisSetCommands#sRandMember(byte[])

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

@@ -99,6 +99,19 @@ class LettuceReactiveSetCommands implements ReactiveSetCommands {
 		}));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sPop(org.springframework.data.redis.connection.ReactiveSetCommands.SPopCommand)
+	 */
+	@Override
+	public Flux<ByteBuffer> sPop(SPopCommand command) {
+
+		Assert.notNull(command, "Command must not be null!");
+		Assert.notNull(command.getKey(), "Key must not be null!");
+
+		return connection.execute(cmd -> cmd.spop(command.getKey(), command.getCount()));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sMove(org.reactivestreams.Publisher)

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

@@ -21,9 +21,11 @@ import io.lettuce.core.ValueScanCursor;
 import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
 import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
+import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.RedisSetCommands;
 import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
@@ -255,6 +257,29 @@ class LettuceSetCommands implements RedisSetCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSetCommands#sPop(byte[], long)
+	 */
+	@Override
+	public List<byte[]> sPop(byte[] key, long count) {
+		try {
+			if (isPipelined()) {
+				pipeline(connection.newLettuceResult(getAsyncConnection().spop(key, count),
+						(Converter<Set<byte[]>, List<byte[]>>) ArrayList::new));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection.newLettuceTxResult(getConnection().spop(key, count),
+						(Converter<Set<byte[]>, List<byte[]>>) ArrayList::new));
+				return null;
+			}
+			return new ArrayList<>(getConnection().spop(key, count));
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisSetCommands#sRandMember(byte[])

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

@@ -106,6 +106,18 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 		return createMono(connection -> connection.sPop(rawKey(key)).map(this::readValue));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveSetOperations#pop(java.lang.Object, long)
+	 */
+	@Override
+	public Flux<V> pop(K key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createFlux(connection -> connection.sPop(rawKey(key), count).map(this::readValue));
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#move(java.lang.Object, java.lang.Object, java.lang.Object)
 	 */

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

@@ -206,6 +206,19 @@ class DefaultSetOperations<K, V> extends AbstractOperations<K, V> implements Set
 		}, true);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.SetOperations(java.lang.Object, long)
+	 */
+	@Override
+	public List<V> pop(K key, long count) {
+
+		final byte[] rawKey = rawKey(key);
+
+		List<byte[]> rawValues = execute(connection -> connection.sPop(rawKey, count), true);
+		return deserializeValues(rawValues);
+	}
+
 	public Long size(K key) {
 		final byte[] rawKey = rawKey(key);
 		return execute(new RedisCallback<Long>() {

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

@@ -59,6 +59,16 @@ public interface ReactiveSetOperations<K, V> {
 	 */
 	Mono<V> pop(K key);
 
+	/**
+	 * Remove and return {@code count} random members from set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count nr of members to return
+	 * @return {@link Flux} emitting random members.
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 */
+	Flux<V> pop(K key, long count);
+
 	/**
 	 * Move {@code value} from {@code key} to {@code destKey}
 	 *

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

@@ -57,6 +57,17 @@ public interface SetOperations<K, V> {
 	 */
 	V pop(K key);
 
+	/**
+	 * Remove and return {@code count} random members from set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count nr of members to return.
+	 * @return empty {@link List} if key does not exist. Never {@literal null}.
+	 * @see <a href="http://redis.io/commands/spop">Redis Documentation: SPOP</a>
+	 * @since 2.0
+	 */
+	List<V> pop(K key, long count);
+
 	/**
 	 * Move {@code value} from {@code key} to {@code destKey}
 	 *