DATAREDIS-693 - Add support for UNLINK.

We now support UNLINK for both Jedis & Lettuce throughout the single node and cluster connection. Reactive support is only available for Lettuce.

Original pull request: #294.

Author: christophstrobl

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

@@ -252,6 +252,15 @@ public Long del(byte[]... keys) {
 		return convertAndReturn(delegate.del(keys), identityConverter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#unlink(byte[][])
+	 */
+	@Override
+	public Long unlink(byte[]... keys) {
+		return convertAndReturn(delegate.unlink(keys), identityConverter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisTxCommands#discard()
@@ -1785,6 +1794,15 @@ public Long del(String... keys) {
 		return del(serializeMulti(keys));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#unlink(java.lang.String[])
+	 */
+	@Override
+	public Long unlink(String... keys) {
+		return unlink(serializeMulti(keys));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#echo(java.lang.String)

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

@@ -68,6 +68,13 @@ default Long del(byte[]... keys) {
 		return keyCommands().del(keys);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#keyCommands()}. */
+	@Override
+	@Deprecated
+	default Long unlink(byte[]... keys) {
+		return keyCommands().unlink(keys);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#keyCommands()}. */
 	@Override
 	@Deprecated

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

@@ -264,6 +264,31 @@ default Mono<Long> del(ByteBuffer key) {
 	 */
 	Flux<NumericResponse<KeyCommand, Long>> del(Publisher<KeyCommand> keys);
 
+	/**
+	 * Unlink {@literal key}.
+	 *
+	 * @param keys 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) {
+
+		Assert.notNull(keys, "Key must not be null!");
+
+		return unlink(Mono.just(keys)).next().map(NumericResponse::getOutput);
+	}
+
+	/**
+	 * Unlink {@literal keys}.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @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>> unlink(Publisher<List<ByteBuffer>> keys);
+
 	/**
 	 * Delete multiple {@literal keys} one in one batch.
 	 *

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

@@ -70,6 +70,18 @@ default Boolean exists(byte[] key) {
 	@Nullable
 	Long del(byte[]... keys);
 
+	/**
+	 * Unlinks the {@code keys} from the keyspace. Unlike with {@link #del(byte[]...)} the actual removal here happens
+	 * asynchronously.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return {@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(byte[]... keys);
+
 	/**
 	 * Determine the type stored at {@code key}.
 	 *

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

@@ -113,6 +113,18 @@ 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.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return {@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(String... keys);
+
 	/**
 	 * Determine the type stored at {@code key}.
 	 *

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

@@ -81,6 +81,20 @@ public Long del(byte[]... keys) {
 				.resultsAsList().size();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#unlink(byte[][])
+	 */
+	@Nullable
+	@Override
+	public Long unlink(byte[]... keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		return connection.<Long> execute("UNLINK", Arrays.asList(keys), Collections.emptyList()).stream()
+				.mapToLong(val -> val).sum();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisKeyCommands#type(byte[])

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

@@ -120,6 +120,19 @@ public Long del(byte[]... keys) {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#unlink(byte[][])
+	 */
+	@Nullable
+	@Override
+	public Long unlink(byte[]... keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		return Long.class.cast(connection.execute("UNLINK", keys));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisKeyCommands#type(byte[])

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

@@ -126,6 +126,30 @@ public Long del(byte[]... keys) {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#unlink(byte[][])
+	 */
+	@Override
+	public Long unlink(byte[]... keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		try {
+			if (isPipelined()) {
+				pipeline(connection.newLettuceResult(getAsyncConnection().unlink(keys)));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection.newLettuceResult(getAsyncConnection().unlink(keys)));
+				return null;
+			}
+			return getConnection().unlink(keys);
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisKeyCommands#type(byte[])

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

@@ -175,6 +175,22 @@ public Flux<NumericResponse<KeyCommand, Long>> del(Publisher<KeyCommand> command
 		}));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#unlink(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<NumericResponse<List<ByteBuffer>, Long>> unlink(Publisher<List<ByteBuffer>> keysCollection) {
+
+		return connection.execute(cmd -> Flux.from(keysCollection).flatMap((keys) -> {
+
+			Assert.notEmpty(keys, "Keys must not be null!");
+
+			return cmd.unlink(keys.stream().collect(Collectors.toList()).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)

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

@@ -2790,6 +2790,26 @@ 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());
 	}

src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java

@@ -2219,4 +2219,18 @@ public void touchReturnsZeroIfNoKeysTouched() {
 		assertThat(clusterConnection.keyCommands().touch(KEY_1_BYTES), is(0L));
 	}
 
+	@Test // DATAREDIS-693
+	public void unlinkReturnsNrOfKeysTouched() {
+
+		nativeConnection.set(KEY_1, VALUE_1);
+		nativeConnection.set(KEY_2, VALUE_1);
+
+		assertThat(clusterConnection.keyCommands().unlink(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES), is(2L));
+	}
+
+	@Test // DATAREDIS-693
+	public void unlinkReturnsZeroIfNoKeysTouched() {
+		assertThat(clusterConnection.keyCommands().unlink(KEY_1_BYTES), is(0L));
+	}
+
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java

@@ -2182,4 +2182,32 @@ public void zUnionStoreShouldWorkForSameSlotKeys() {
 
 		assertThat(nativeConnection.zrange(SAME_SLOT_KEY_3, 0, -1), hasItems(VALUE_1, VALUE_2, VALUE_3));
 	}
+
+	@Test // DATAREDIS-694
+	public void touchReturnsNrOfKeysTouched() {
+
+		nativeConnection.set(KEY_1, VALUE_1);
+		nativeConnection.set(KEY_2, VALUE_1);
+
+		assertThat(clusterConnection.keyCommands().touch(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES), is(2L));
+	}
+
+	@Test // DATAREDIS-694
+	public void touchReturnsZeroIfNoKeysTouched() {
+		assertThat(clusterConnection.keyCommands().touch(KEY_1_BYTES), is(0L));
+	}
+
+	@Test // DATAREDIS-693
+	public void unlinkReturnsNrOfKeysTouched() {
+
+		nativeConnection.set(KEY_1, VALUE_1);
+		nativeConnection.set(KEY_2, VALUE_1);
+
+		assertThat(clusterConnection.keyCommands().unlink(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES), is(2L));
+	}
+
+	@Test // DATAREDIS-693
+	public void unlinkReturnsZeroIfNoKeysTouched() {
+		assertThat(clusterConnection.keyCommands().unlink(KEY_1_BYTES), is(0L));
+	}
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommandsTests.java

@@ -29,11 +29,14 @@
 import java.time.Instant;
 import java.util.Arrays;
 
+import org.junit.Rule;
 import org.junit.Test;
 import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
+import org.springframework.data.redis.test.util.MinimumRedisVersionRule;
+import org.springframework.test.annotation.IfProfileValue;
 
 /**
  * Integration tests for {@link LettuceReactiveKeyCommands}.
@@ -43,6 +46,8 @@
  */
 public class LettuceReactiveKeyCommandsTests extends LettuceReactiveCommandsTestsBase {
 
+	public @Rule MinimumRedisVersionRule versionRule = new MinimumRedisVersionRule();
+
 	@Test // DATAREDIS-525
 	public void existsShouldReturnTrueForExistingKeys() {
 
@@ -306,4 +311,26 @@ public void touchReturnsZeroIfNoKeysTouched() {
 				.expectNext(0L) //
 				.verifyComplete();
 	}
+
+	@Test // DATAREDIS-693
+	@IfProfileValue(name = "redisVersion", value = "4.0.0+")
+	public void unlinkReturnsNrOfKeysRemoved() {
+
+		nativeCommands.set(KEY_1, VALUE_1);
+		nativeCommands.set(KEY_2, VALUE_2);
+
+		StepVerifier.create(connection.keyCommands().unlink(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER, KEY_3_BBUFFER)))
+				.expectNext(2L) //
+				.verifyComplete();
+
+	}
+
+	@Test // DATAREDIS-693
+	@IfProfileValue(name = "redisVersion", value = "4.0.0+")
+	public void unlinkReturnsZeroIfNoKeysRemoved() {
+
+		StepVerifier.create(connection.keyCommands().unlink(Arrays.asList(KEY_1_BBUFFER))) //
+				.expectNext(0L) //
+				.verifyComplete();
+	}
 }