DATAREDIS-693 - Polishing.

Add Template methods for unlink(…). Align unlink with del/mDel methods on reactive API using concatMap to retain order and provide batching/streaming. Reword Javadoc. Extend tests.

Replace blocking calls with StepVerifier in LettuceReactiveKeyCommandsTests.

Original pull request: #294.

Author: mp911de

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

@@ -265,52 +265,81 @@ default Mono<Long> del(ByteBuffer key) {
 	Flux<NumericResponse<KeyCommand, Long>> del(Publisher<KeyCommand> keys);
 
 	/**
-	 * Unlink {@literal key}.
+	 * Delete multiple {@literal keys} one in one batch.
 	 *
 	 * @param keys must not be {@literal null}.
 	 * @return
+	 * @see <a href="http://redis.io/commands/del">Redis Documentation: DEL</a>
+	 */
+	default Mono<Long> mDel(List<ByteBuffer> keys) {
+
+		Assert.notEmpty(keys, "Keys must not be empty or null!");
+
+		return mDel(Mono.just(keys)).next().map(NumericResponse::getOutput);
+	}
+
+	/**
+	 * Delete multiple {@literal keys} in batches.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return {@link Flux} of {@link NumericResponse} holding the {@literal keys} removed along with the deletion result.
+	 * @see <a href="http://redis.io/commands/del">Redis Documentation: DEL</a>
+	 */
+	Flux<NumericResponse<List<ByteBuffer>, Long>> mDel(Publisher<List<ByteBuffer>> keys);
+
+	/**
+	 * Unlink the {@code key} from the keyspace. Unlike with {@link #del(ByteBuffer)} the actual memory reclaiming here
+	 * happens asynchronously.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
 	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
 	 * @since 2.1
 	 */
-	default Mono<Long> unlink(List<ByteBuffer> keys) {
+	default Mono<Long> unlink(ByteBuffer key) {
 
-		Assert.notNull(keys, "Key must not be null!");
+		Assert.notNull(key, "Keys must not be null!");
 
-		return unlink(Mono.just(keys)).next().map(NumericResponse::getOutput);
+		return unlink(Mono.just(key).map(KeyCommand::new)).next().map(NumericResponse::getOutput);
 	}
 
 	/**
-	 * Unlink {@literal keys}.
+	 * Unlink the {@code key} from the keyspace. Unlike with {@link #del(ByteBuffer)} the actual memory reclaiming here
+	 * happens asynchronously.
 	 *
 	 * @param keys must not be {@literal null}.
-	 * @return {@link Flux} of {@link NumericResponse} holding the {@literal key} removed along with the deletion result.
+	 * @return {@link Flux} of {@link NumericResponse} holding the {@literal key} removed along with the unlink result.
 	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
 	 * @since 2.1
 	 */
-	Flux<NumericResponse<List<ByteBuffer>, Long>> unlink(Publisher<List<ByteBuffer>> keys);
+	Flux<NumericResponse<KeyCommand, Long>> unlink(Publisher<KeyCommand> keys);
 
 	/**
-	 * Delete multiple {@literal keys} one in one batch.
+	 * Unlink the {@code keys} from the keyspace. Unlike with {@link #mDel(List)} the actual memory reclaiming here
+	 * happens asynchronously.
 	 *
 	 * @param keys must not be {@literal null}.
 	 * @return
-	 * @see <a href="http://redis.io/commands/del">Redis Documentation: DEL</a>
+	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
+	 * @since 2.1
 	 */
-	default Mono<Long> mDel(List<ByteBuffer> keys) {
+	default Mono<Long> mUnlink(List<ByteBuffer> keys) {
 
-		Assert.notEmpty(keys, "Keys must not be empty or null!");
+		Assert.notNull(keys, "Keys must not be null!");
 
-		return mDel(Mono.just(keys)).next().map(NumericResponse::getOutput);
+		return mUnlink(Mono.just(keys)).next().map(NumericResponse::getOutput);
 	}
 
 	/**
-	 * Delete multiple {@literal keys} in batches.
+	 * Unlink the {@code keys} from the keyspace. Unlike with {@link #mDel(Publisher)} the actual memory reclaiming here
+	 * happens asynchronously.
 	 *
 	 * @param keys must not be {@literal null}.
-	 * @return {@link Flux} of {@link NumericResponse} holding the {@literal keys} removed along with the deletion result.
-	 * @see <a href="http://redis.io/commands/del">Redis Documentation: DEL</a>
+	 * @return {@link Flux} of {@link NumericResponse} holding the {@literal key} removed along with the deletion result.
+	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
+	 * @since 2.1
 	 */
-	Flux<NumericResponse<List<ByteBuffer>, Long>> mDel(Publisher<List<ByteBuffer>> keys);
+	Flux<NumericResponse<List<ByteBuffer>, Long>> mUnlink(Publisher<List<ByteBuffer>> keys);
 
 	/**
 	 * {@code EXPIRE}/{@code PEXPIRE} command parameters.

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

@@ -71,8 +71,8 @@ default Boolean exists(byte[] key) {
 	Long del(byte[]... keys);
 
 	/**
-	 * Unlinks the {@code keys} from the keyspace. Unlike with {@link #del(byte[]...)} the actual removal here happens
-	 * asynchronously.
+	 * Unlink the {@code keys} from the keyspace. Unlike with {@link #del(byte[]...)} the actual memory reclaiming here
+	 * happens asynchronously.
 	 *
 	 * @param keys must not be {@literal null}.
 	 * @return {@literal null} when used in pipeline / transaction.

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

@@ -114,8 +114,8 @@ interface StringTuple extends Tuple {
 	Long del(String... keys);
 
 	/**
-	 * Unlinks the {@code keys} from the keyspace. Unlike with {@link #del(byte[]...)} the actual removal here happens
-	 * asynchronously.
+	 * Unlink the {@code keys} from the keyspace. Unlike with {@link #del(String...)} the actual memory reclaiming here
+	 * happens asynchronously.
 	 *
 	 * @param keys must not be {@literal null}.
 	 * @return {@literal null} when used in pipeline / transaction.

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

@@ -22,7 +22,6 @@
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.redis.connection.DataType;
@@ -177,33 +176,47 @@ public Flux<NumericResponse<KeyCommand, Long>> del(Publisher<KeyCommand> command
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#unlink(org.reactivestreams.Publisher)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#mDel(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<List<ByteBuffer>, Long>> unlink(Publisher<List<ByteBuffer>> keysCollection) {
+	public Flux<NumericResponse<List<ByteBuffer>, Long>> mDel(Publisher<List<ByteBuffer>> keysCollection) {
 
-		return connection.execute(cmd -> Flux.from(keysCollection).flatMap((keys) -> {
+		return connection.execute(cmd -> Flux.from(keysCollection).concatMap((keys) -> {
 
 			Assert.notEmpty(keys, "Keys must not be null!");
 
-			return cmd.unlink(keys.stream().collect(Collectors.toList()).toArray(new ByteBuffer[keys.size()]))
+			return cmd.del(keys.toArray(new ByteBuffer[keys.size()]))
 					.map((value) -> new NumericResponse<>(keys, value));
 		}));
 	}
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#mDel(org.reactivestreams.Publisher)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#unlink(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<List<ByteBuffer>, Long>> mDel(Publisher<List<ByteBuffer>> keysCollection) {
+	public Flux<NumericResponse<KeyCommand, Long>> unlink(Publisher<KeyCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).concatMap((command) -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+
+			return cmd.unlink(command.getKey()).map((value) -> new NumericResponse<>(command, value));
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#mUnlink(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<NumericResponse<List<ByteBuffer>, Long>> mUnlink(Publisher<List<ByteBuffer>> keysCollection) {
 
 		return connection.execute(cmd -> Flux.from(keysCollection).concatMap((keys) -> {
 
 			Assert.notEmpty(keys, "Keys must not be null!");
 
-			return cmd.del(keys.stream().collect(Collectors.toList()).toArray(new ByteBuffer[keys.size()]))
-					.map((value) -> new NumericResponse<>(keys, value));
+			return cmd.unlink(keys.toArray(new ByteBuffer[keys.size()])).map((value) -> new NumericResponse<>(keys, value));
 		}));
 	}
 

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

@@ -132,6 +132,28 @@ public interface ReactiveRedisOperations<K, V> {
 	 */
 	Mono<Long> delete(Publisher<K> keys);
 
+	/**
+	 * Unlink the {@code key} from the keyspace. Unlike with {@link #delete(Object[])} the actual memory reclaiming here
+	 * happens asynchronously.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
+	 * @since 2.1
+	 */
+	Mono<Long> unlink(K... key);
+
+	/**
+	 * Unlink the {@code keys} from the keyspace. Unlike with {@link #delete(Publisher)} the actual memory reclaiming here
+	 * happens asynchronously.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
+	 * @since 2.1
+	 */
+	Mono<Long> unlink(Publisher<K> keys);
+
 	/**
 	 * Set time to live for given {@code key}.
 	 *

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

@@ -304,6 +304,40 @@ public Mono<Long> delete(Publisher<K> keys) {
 				.map(CommandResponse::getOutput));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#unlink(java.lang.Object[])
+	 */
+	@Override
+	@SafeVarargs
+	public final Mono<Long> unlink(K... keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+		Assert.notEmpty(keys, "Keys must not be empty!");
+		Assert.noNullElements(keys, "Keys must not contain null elements!");
+
+		if (keys.length == 1) {
+			return createMono(connection -> connection.keyCommands().unlink(rawKey(keys[0])));
+		}
+
+		Mono<List<ByteBuffer>> listOfKeys = Flux.fromArray(keys).map(this::rawKey).collectList();
+		return createMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mUnlink(rawKeys)));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#unlink(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Mono<Long> unlink(Publisher<K> keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		return createMono(connection -> connection.keyCommands() //
+				.unlink(Flux.from(keys).map(this::rawKey).map(KeyCommand::new)) //
+				.map(CommandResponse::getOutput));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#expire(java.lang.Object, java.time.Duration)

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

@@ -194,6 +194,30 @@ <T> T execute(RedisScript<T> script, RedisSerializer<?> argsSerializer, RedisSer
 	@Nullable
 	Long delete(Collection<K> keys);
 
+	/**
+	 * Unlink the {@code key} from the keyspace. Unlike with {@link #delete(Object)} the actual memory reclaiming here
+	 * happens asynchronously.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
+	 * @since 2.1
+	 */
+	@Nullable
+	Boolean unlink(K key);
+
+	/**
+	 * Unlink the {@code keys} from the keyspace. Unlike with {@link #delete(Collection)} the actual memory reclaiming
+	 * here happens asynchronously.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="http://redis.io/commands/unlink">Redis Documentation: UNLINK</a>
+	 * @since 2.1
+	 */
+	@Nullable
+	Long unlink(Collection<K> keys);
+
 	/**
 	 * Determine the type stored at {@code key}.
 	 *

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

@@ -727,6 +727,36 @@ public Long delete(Collection<K> keys) {
 		return execute(connection -> connection.del(rawKeys), true);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisOperations#unlink(java.lang.Object)
+	 */
+	@Override
+	public Boolean unlink(K key) {
+
+		byte[] rawKey = rawKey(key);
+
+		Long result = execute(connection -> connection.unlink(rawKey), true);
+
+		return result != null && result.intValue() == 1;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisOperations#unlink(java.util.Collection)
+	 */
+	@Override
+	public Long unlink(Collection<K> keys) {
+
+		if (CollectionUtils.isEmpty(keys)) {
+			return 0L;
+		}
+
+		byte[][] rawKeys = rawKeys(keys);
+
+		return execute(connection -> connection.unlink(rawKeys), true);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.RedisOperations#hasKey(java.lang.Object)

src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java

@@ -1128,6 +1128,38 @@ public void testDel() {
 		verifyResults(Arrays.asList(true, 1L, false));
 	}
 
+	@Test // DATAREDIS-693
+	@IfProfileValue(name = "redisVersion", value = "4.0+")
+	public void unlinkReturnsNrOfKeysRemoved() {
+
+		connection.set("unlink.this", "Can't track this!");
+
+		actual.add(connection.unlink("unlink.this", "unlink.that"));
+
+		verifyResults(Arrays.asList(new Object[] { 1L }));
+	}
+
+	@Test // DATAREDIS-693
+	@IfProfileValue(name = "redisVersion", value = "4.0+")
+	public void testUnlinkBatch() {
+
+		actual.add(connection.set("testing", "123"));
+		actual.add(connection.set("foo", "bar"));
+		actual.add(connection.unlink("testing", "foo"));
+		actual.add(connection.exists("testing"));
+
+		verifyResults(Arrays.asList(true, true, 2L, false));
+	}
+
+	@Test // DATAREDIS-693
+	@IfProfileValue(name = "redisVersion", value = "4.0+")
+	public void unlinkReturnsZeroIfNoKeysRemoved() {
+
+		actual.add(connection.unlink("unlink.this"));
+
+		verifyResults(Arrays.asList(new Object[] { 0L }));
+	}
+
 	@Test
 	public void testType() {
 
@@ -2790,26 +2822,6 @@ public void touchReturnsZeroIfNoKeysTouched() {
 		verifyResults(Arrays.asList(new Object[] { 0L }));
 	}
 
-	@Test // DATAREDIS-693
-	@IfProfileValue(name = "redisVersion", value = "4.0+")
-	public void unlinkReturnsNrOfKeysRemoved() {
-
-		connection.set("unlink.this", "Can't track this!");
-
-		actual.add(connection.unlink("unlink.this", "unlink.that"));
-
-		verifyResults(Arrays.asList(new Object[] { 1L }));
-	}
-
-	@Test // DATAREDIS-693
-	@IfProfileValue(name = "redisVersion", value = "4.0+")
-	public void unlinkReturnsZeroIfNoKeysRemoved() {
-
-		actual.add(connection.unlink("unlink.this"));
-
-		verifyResults(Arrays.asList(new Object[] { 0L }));
-	}
-
 	protected void verifyResults(List<Object> expected) {
 		assertEquals(expected, getResults());
 	}