Add exists(…) accepting multiple keys to Reactive API.

Closes #2883
Original pull request: #2918

Signed-off-by: Anne Lee <anne.lee@buzzvil.com>

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

@@ -177,6 +177,14 @@ public interface ReactiveKeyCommands {
 
 		return exists(Mono.just(new KeyCommand(key))).next().map(BooleanResponse::getOutput);
 	}
+	/**
+	 * Determine the number of given {@literal keys} that exist.
+	 *
+	 * @param keys must not be {@literal null} or {@literal empty}.
+	 * @return {@link Mono} emitting {@literal the number of existing keys}.
+	 * @see <a href="https://redis.io/docs/commands/exists/">Redis Documentation: EXISTS</a>
+	 */
+	Mono<Long> exists(List<ByteBuffer> keys);
 
 	/**
 	 * Determine if given {@literal key} exists.

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

@@ -344,6 +344,14 @@ class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
 		return connection.execute(cmd -> cmd.objectRefcount(key)).next();
 	}
 
+	@Override
+	public Mono<Long> exists(List<ByteBuffer> keys) {
+		Assert.notNull(keys, "Key list must not be null");
+		Assert.notEmpty(keys, "Key list must not be empty");
+
+		return connection.execute(cmd -> cmd.exists(keys.toArray(ByteBuffer[]::new))).next();
+	}
+
 	private static ExpireArgs getExpireArgs(ExpirationOptions options) {
 
 		return new ExpireArgs() {

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

@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
@@ -421,6 +422,15 @@ public interface ReactiveRedisOperations<K, V> {
 	 */
 	Mono<Duration> getExpire(K key);
 
+	/**
+	 * Get the number of given {@code keys} that exists.
+	 *
+	 * @param keys must not be {@literal null} or {@literal empty}.
+	 * @return the number of existing keys in redis. 0 if there are no existing keys.
+	 * @see <a href="https://redis.io/docs/commands/exists/">Redis Documentation: EXISTS</a>
+	 */
+	Mono<Long> countExistingKeys(Collection<K> keys);
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with Redis Lua scripts
 	// -------------------------------------------------------------------------

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

@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -505,6 +506,14 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return doCreateMono(connection -> connection.keyCommands().move(rawKey(key), dbIndex));
 	}
 
+	@Override
+	public Mono<Long> countExistingKeys(Collection<K> keys) {
+		Assert.notNull(keys, "Keys must not be null");
+
+		ByteBuffer[] rawKeys = rawKeys(keys);
+		return doCreateMono(connection -> connection.keyCommands().exists(Arrays.asList(rawKeys)));
+	}
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with Redis Lua scripts
 	// -------------------------------------------------------------------------
@@ -688,6 +697,17 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return getSerializationContext().getKeySerializationPair().getWriter().write(key);
 	}
 
+	private ByteBuffer[] rawKeys(Collection<K> keys) {
+		final ByteBuffer[] rawKeys = new ByteBuffer[keys.size()];
+
+		int i = 0;
+		for (K key : keys) {
+			rawKeys[i++] = rawKey(key);
+		}
+
+		return rawKeys;
+	}
+
 	@Nullable
 	private K readKey(ByteBuffer buffer) {
 		return getSerializationContext().getKeySerializationPair().getReader().read(buffer);

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

@@ -70,6 +70,22 @@ public class LettuceReactiveKeyCommandsIntegrationTests extends LettuceReactiveC
 		connection.keyCommands().exists(KEY_1_BBUFFER).as(StepVerifier::create).expectNext(false).verifyComplete();
 	}
 
+	@ParameterizedRedisTest
+	void existsKeyReturnsKeyCount() {
+		nativeCommands.set(KEY_1, "1000");
+		nativeCommands.set(KEY_2, "2000");
+		nativeCommands.set(KEY_3, "3000");
+
+		connection.keyCommands().exists(List.of(KEY_1_BBUFFER, KEY_2_BBUFFER, KEY_3_BBUFFER)).as(StepVerifier::create)
+				.expectNext(3L).verifyComplete();
+	}
+
+	@ParameterizedRedisTest
+	void existsKeyReturnsZeroWhenKeyDoesNotExist() {
+		connection.keyCommands().exists(List.of(KEY_1_BBUFFER, KEY_2_BBUFFER, KEY_3_BBUFFER)).as(StepVerifier::create)
+				.expectNext(0L).verifyComplete();
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-525
 	void typeShouldReturnTypeCorrectly() {
 
@@ -168,7 +184,7 @@ public class LettuceReactiveKeyCommandsIntegrationTests extends LettuceReactiveC
 		connection.keyCommands().rename(KEY_1_BBUFFER, KEY_2_BBUFFER).as(StepVerifier::create).expectNext(true)
 				.verifyComplete();
 		assertThat(nativeCommands.exists(KEY_2)).isEqualTo(1L);
-		assertThat(nativeCommands.exists(KEY_1)).isEqualTo(0L);
+		assertThat(nativeCommands.exists(KEY_1)).isZero();
 	}
 
 	@ParameterizedRedisTest // DATAREDIS-525
@@ -187,7 +203,7 @@ public class LettuceReactiveKeyCommandsIntegrationTests extends LettuceReactiveC
 				.verifyComplete();
 
 		assertThat(nativeCommands.exists(KEY_2)).isEqualTo(1L);
-		assertThat(nativeCommands.exists(KEY_1)).isEqualTo(0L);
+		assertThat(nativeCommands.exists(KEY_1)).isZero();
 	}
 
 	@ParameterizedRedisTest // DATAREDIS-525
@@ -428,7 +444,7 @@ public class LettuceReactiveKeyCommandsIntegrationTests extends LettuceReactiveC
 				.expectNext(true) //
 				.expectComplete() //
 				.verify();
-		assertThat(nativeCommands.exists(KEY_1)).isEqualTo(0L);
+		assertThat(nativeCommands.exists(KEY_1)).isZero();
 	}
 
 	@ParameterizedRedisTest // DATAREDIS-694

src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java

@@ -24,10 +24,7 @@ import reactor.test.StepVerifier;
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 import java.util.function.Function;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -582,4 +579,24 @@ public class ReactiveRedisTemplateIntegrationTests<K, V> {
 				.thenCancel() //
 				.verify(Duration.ofSeconds(3));
 	}
+
+	@ParameterizedRedisTest
+	void countExistingKeysIfValidKeyExists() {
+
+		K key = keyFactory.instance();
+		K key2 = keyFactory.instance();
+		K key3 = keyFactory.instance();
+
+		redisTemplate.opsForValue().set(key, valueFactory.instance()).as(StepVerifier::create).expectNext(true).verifyComplete();
+		redisTemplate.opsForValue().set(key2, valueFactory.instance()).as(StepVerifier::create).expectNext(true).verifyComplete();
+		redisTemplate.opsForValue().set(key3, valueFactory.instance()).as(StepVerifier::create).expectNext(true).verifyComplete();
+
+		redisTemplate.countExistingKeys(Arrays.asList(key, key2, key3)).as(StepVerifier::create).expectNext(3L).verifyComplete();
+	}
+
+	@ParameterizedRedisTest
+	void countExistingKeysIfNotValidKeyExists() {
+		K key = keyFactory.instance();
+		redisTemplate.countExistingKeys(List.of(key)).as(StepVerifier::create).expectNext(0L).verifyComplete();
+	}
 }