DATAREDIS-529 - Add support for EXISTS taking multiple keys.

Original pull request: #281.

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

@@ -15,17 +15,8 @@
  */
 package org.springframework.data.redis.connection;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Queue;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
@@ -303,6 +294,24 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.exists(key), identityConverter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#exists(String[])
+	 */
+	@Override
+	public Long exists(String... keys) {
+		return convertAndReturn(delegate.exists(Arrays.asList(serializeMulti(keys))), identityConverter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#exists(java.util.Collection)
+	 */
+	@Override
+	public Long exists(Collection<byte[]> keys) {
+		return convertAndReturn(delegate.exists(keys), identityConverter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisKeyCommands#expire(byte[], long)

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.connection;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -54,6 +55,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return keyCommands().exists(key);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#keyCommands()}. */
+	@Override
+	@Deprecated
+	default Long exists(Collection<byte[]> keys) {
+		return keyCommands().exists(keys);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#keyCommands()}. */
 	@Override
 	@Deprecated

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.redis.connection;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -22,6 +24,7 @@ import java.util.concurrent.TimeUnit;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * Key-specific commands supported by Redis.
@@ -40,7 +43,23 @@ public interface RedisKeyCommands {
 	 * @see <a href="http://redis.io/commands/exists">Redis Documentation: EXISTS</a>
 	 */
 	@Nullable
-	Boolean exists(byte[] key);
+	default Boolean exists(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Long count = exists(Collections.singleton(key));
+		return count != null ? count > 0 : null;
+	}
+
+	/**
+	 * Count how many of the given {@code keys} exists. Providing the very same {@code key} more than once also counts
+	 * multiple times.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return the number of keys existing among the ones specified as arguments.
+	 * @since 2.1
+	 */
+	@Nullable
+	Long exists(Collection<byte[]> keys);
 
 	/**
 	 * Delete given {@code keys}.

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

@@ -91,6 +91,18 @@ public interface StringRedisConnection extends RedisConnection {
 	 */
 	Boolean exists(String key);
 
+	/**
+	 * Count how many of the if given {@code keys} exists.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/exists">Redis Documentation: EXISTS</a>
+	 * @see RedisKeyCommands#exists(java.util.Collection)
+	 * @since 2.1
+	 */
+	@Nullable
+	Long exists(String... keys);
+
 	/**
 	 * Delete given {@code keys}.
 	 *

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

@@ -37,8 +37,10 @@ import org.springframework.data.redis.connection.jedis.JedisClusterConnection.Je
 import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisMultiKeyClusterCommandCallback;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.ObjectUtils;
 
 /**
  * @author Christoph Strobl
@@ -471,15 +473,23 @@ class JedisClusterKeyCommands implements RedisKeyCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.RedisKeyCommands#exists(byte[])
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#exists(java.util.Collection)
 	 */
+	@Nullable
 	@Override
-	public Boolean exists(byte[] key) {
+	public Long exists(Collection<byte[]> keys) {
 
-		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(keys, "Keys must not be null!");
 
 		try {
-			return connection.getCluster().exists(key);
+
+			return keys.stream() //
+					.parallel()
+					.map(key -> connection.getClusterCommandExecutor()
+							.executeCommandOnSingleNode((JedisClusterCommandCallback<Boolean>) client -> client.exists(key),
+									connection.getTopologyProvider().getTopology().getKeyServingMasterNode(key))
+							.getValue())
+					.mapToLong(val -> ObjectUtils.nullSafeEquals(val, Boolean.TRUE) ? 1L : 0L).sum();
 		} catch (Exception ex) {
 			throw convertJedisAccessException(ex);
 		}

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

@@ -20,6 +20,7 @@ import lombok.RequiredArgsConstructor;
 import redis.clients.jedis.ScanParams;
 import redis.clients.jedis.SortingParams;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -70,6 +71,33 @@ class JedisKeyCommands implements RedisKeyCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#exists(java.util.Collection)
+	 */
+	@Nullable
+	@Override
+	public Long exists(Collection<byte[]> keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		try {
+			if (isPipelined()) {
+				pipeline(
+						connection.newJedisResult(connection.getRequiredPipeline().exists(keys.toArray(new byte[keys.size()][]))));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection
+						.newJedisResult(connection.getRequiredTransaction().exists(keys.toArray(new byte[keys.size()][]))));
+				return null;
+			}
+			return connection.getJedis().exists(keys.toArray(new byte[keys.size()][]));
+		} catch (Exception ex) {
+			throw connection.convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisKeyCommands#del(byte[][])

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

@@ -23,6 +23,7 @@ import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -38,6 +39,7 @@ import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.ScanCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -76,6 +78,31 @@ class LettuceKeyCommands implements RedisKeyCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#exists(java.util.Collection)
+	 */
+	@Nullable
+	@Override
+	public Long exists(Collection<byte[]> keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		try {
+			if (isPipelined()) {
+				pipeline(connection.newLettuceResult(getAsyncConnection().exists(keys.toArray(new byte[keys.size()][]))));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection.newLettuceTxResult(getAsyncConnection().exists(keys.toArray(new byte[keys.size()][]))));
+				return null;
+			}
+			return getConnection().exists(keys.toArray(new byte[keys.size()][]));
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisKeyCommands#del(byte[][])

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

@@ -162,6 +162,18 @@ public interface RedisOperations<K, V> {
 	@Nullable
 	Boolean hasKey(K key);
 
+	/**
+	 * Count the number of {@code keys} that exists.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return The number of keys existing among the ones specified as arguments. Keys mentioned multiple times and
+	 *         existing are counted multiple times.
+	 * @see <a href="http://redis.io/commands/exists">Redis Documentation: EXISTS</a>
+	 * @since 2.1
+	 */
+	@Nullable
+	Long countExistingKeys(Collection<K> keys);
+
 	/**
 	 * Delete given {@code key}.
 	 *

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

@@ -18,6 +18,7 @@ package org.springframework.data.redis.core;
 import java.io.Closeable;
 import java.lang.reflect.Proxy;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
@@ -739,6 +740,19 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return execute(connection -> connection.exists(rawKey), true);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisOperations#countExistingKeys(java.util.Collection)
+	 */
+	@Override
+	public Long countExistingKeys(Collection<K> keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		byte[][] rawKeys = rawKeys(keys);
+		return execute(connection -> connection.exists(Arrays.asList(rawKeys)), true);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.RedisOperations#expire(java.lang.Object, long, java.util.concurrent.TimeUnit)

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

@@ -908,6 +908,36 @@ public abstract class AbstractConnectionIntegrationTests {
 		verifyResults(Arrays.asList(new Object[] { true, false }));
 	}
 
+	@Test // DATAREDIS-529
+	public void testExistsWithMultipleKeys() {
+
+		connection.set("exist-1", "true");
+		connection.set("exist-2", "true");
+		connection.set("exist-3", "true");
+
+		actual.add(connection.exists("exist-1", "exist-2", "exist-3", "nonexistent"));
+
+		verifyResults(Arrays.asList(new Object[] { 3L }));
+	}
+
+	@Test // DATAREDIS-529
+	public void testExistsWithMultipleKeysNoneExists() {
+
+		actual.add(connection.exists("no-exist-1", "no-exist-2"));
+
+		verifyResults(Arrays.asList(new Object[] { 0L }));
+	}
+
+	@Test // DATAREDIS-529
+	public void testExistsSameKeyMultipleTimes() {
+
+		connection.set("existent", "true");
+
+		actual.add(connection.exists("existent", "existent"));
+
+		verifyResults(Arrays.asList(new Object[] { 2L }));
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testKeys() throws Exception {

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

@@ -23,9 +23,9 @@ import org.springframework.data.geo.Point;
  */
 public interface ClusterConnectionTests {
 
-	static final Point POINT_ARIGENTO = new Point(13.583333, 37.316667);
-	static final Point POINT_CATANIA = new Point(15.087269, 37.502669);
-	static final Point POINT_PALERMO = new Point(13.361389, 38.115556);
+	Point POINT_ARIGENTO = new Point(13.583333, 37.316667);
+	Point POINT_CATANIA = new Point(15.087269, 37.502669);
+	Point POINT_PALERMO = new Point(13.361389, 38.115556);
 
 	// DATAREDIS-315
 	void shouldAllowSettingAndGettingValues();
@@ -642,4 +642,14 @@ public interface ClusterConnectionTests {
 
 	// DATAREDIS-698
 	void hStrLenReturnsZeroWhenKeyDoesNotExist();
+
+	// DATAREDIS-529
+	void testExistsWithMultipleKeys();
+
+	// DATAREDIS-529
+	void testExistsWithMultipleKeysNoneExists();
+
+	// DATAREDIS-529
+	void testExistsSameKeyMultipleTimes();
+
 }

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

@@ -2137,6 +2137,32 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests {
 		assertThat(clusterConnection.geoRemove(KEY_1_BYTES, ARIGENTO.getName()), is(1L));
 	}
 
+	@Test // DATAREDIS-529
+	public void testExistsWithMultipleKeys() {
+
+		nativeConnection.set(KEY_1, "true");
+		nativeConnection.set(KEY_2, "true");
+		nativeConnection.set(KEY_3, "true");
+
+		assertThat(clusterConnection.keyCommands()
+				.exists(Arrays.asList(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES, "nonexistent".getBytes())), is(3L));
+	}
+
+	@Test // DATAREDIS-529
+	public void testExistsWithMultipleKeysNoneExists() {
+
+		assertThat(clusterConnection.keyCommands().exists(Arrays.asList("no-exist-1".getBytes(), "no-exist-2".getBytes())),
+				is(0L));
+	}
+
+	@Test // DATAREDIS-529
+	public void testExistsSameKeyMultipleTimes() {
+
+		nativeConnection.set(KEY_1, "true");
+
+		assertThat(clusterConnection.keyCommands().exists(Arrays.asList(KEY_1_BYTES, KEY_1_BYTES)), is(2L));
+	}
+
 	@Test(expected = IllegalArgumentException.class) // DATAREDIS-689
 	public void executeWithNoKeyAndArgsThrowsException() {
 		clusterConnection.execute("KEYS", null, Collections.singletonList("*".getBytes()));

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

@@ -124,7 +124,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
 	@After
 	public void tearDown() throws InterruptedException {
 
-		clusterConnection.flushDb();
+		clusterConnection.serverCommands().flushDb();
 		nativeConnection.getStatefulConnection().close();
 		clusterConnection.close();
 		client.shutdown(0, 0, TimeUnit.MILLISECONDS);
@@ -2146,6 +2146,32 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
 		assertThat(clusterConnection.geoRemove(KEY_1_BYTES, ARIGENTO_BYTES.getName()), is(1L));
 	}
 
+	@Test // DATAREDIS-529
+	public void testExistsWithMultipleKeys() {
+
+		nativeConnection.set(KEY_1, "true");
+		nativeConnection.set(KEY_2, "true");
+		nativeConnection.set(KEY_3, "true");
+
+		assertThat(clusterConnection.keyCommands()
+				.exists(Arrays.asList(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES, "nonexistent".getBytes())), is(3L));
+	}
+
+	@Test // DATAREDIS-529
+	public void testExistsWithMultipleKeysNoneExists() {
+
+		assertThat(clusterConnection.keyCommands().exists(Arrays.asList("no-exist-1".getBytes(), "no-exist-2".getBytes())),
+				is(0L));
+	}
+
+	@Test // DATAREDIS-529
+	public void testExistsSameKeyMultipleTimes() {
+
+		nativeConnection.set(KEY_1, "true");
+
+		assertThat(clusterConnection.keyCommands().exists(Arrays.asList(KEY_1_BYTES, KEY_1_BYTES)), is(2L));
+	}
+
 	@Test // DATAREDIS-698
 	public void hStrLenReturnsFieldLength() {
 

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

@@ -809,6 +809,19 @@ public class RedisTemplateTests<K, V> {
 		assertThat(redisTemplate.getClientList().size(), IsNot.not(0));
 	}
 
+	@Test // DATAREDIS-529
+	public void countExistingKeysReturnsNumberOfKeysCorrectly() {
+
+		Map<K, V> source = new LinkedHashMap<>(3, 1);
+		source.put(keyFactory.instance(), valueFactory.instance());
+		source.put(keyFactory.instance(), valueFactory.instance());
+		source.put(keyFactory.instance(), valueFactory.instance());
+
+		redisTemplate.opsForValue().multiSet(source);
+
+		assertThat(redisTemplate.countExistingKeys(source.keySet()), is(3L));
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	private byte[] serialize(Object value, RedisSerializer serializer) {
 		if (serializer == null && value instanceof byte[]) {