Add UNLINK as .KeyDelete() automatic usage underneath (#910)

* Add UNLINK as .KeyUnlink()

Submitting as a PR so test as reuable, but maybe we want this as an overload via some enum on .KeyDelete() instead? I'm good either way, but getting the ball rolling on #909 here.

* Add feature check for UNLINK

Ah right, 4.0.0+ only...

* Move UNLINK behind KeyDelete() and always call it if available

We still weren't testing Delete(Async)() in general, so tests added there. I've confirmed this is running DEL against 3.x and UNLINK against 5.x.

* Move GetDeleteCommand after the methods (consistent with rest of the file)

StackExchange.Redis.Tests/BasicOps.cs

@@ -425,6 +425,82 @@ private void Incr(IDatabase database, RedisKey key, int delta, ref int total)
             total += delta;
         }
 
+        [Fact]
+        public async Task Delete()
+        {
+            using (var muxer = Create())
+            {
+                var db = muxer.GetDatabase();
+                var key = Me();
+                var ss = db.StringSetAsync(key, "Heyyyyy");
+                var ke1 = db.KeyExistsAsync(key).ForAwait();
+                var ku1 = db.KeyDelete(key);
+                var ke2 = db.KeyExistsAsync(key).ForAwait();
+                Assert.True(await ke1);
+                Assert.True(ku1);
+                Assert.False(await ke2);
+            }
+        }
+
+        [Fact]
+        public async Task DeleteAsync()
+        {
+            using (var muxer = Create())
+            {
+                var db = muxer.GetDatabase();
+                var key = Me();
+                var ss = db.StringSetAsync(key, "Heyyyyy");
+                var ke1 = db.KeyExistsAsync(key).ForAwait();
+                var ku1 = db.KeyDeleteAsync(key).ForAwait();
+                var ke2 = db.KeyExistsAsync(key).ForAwait();
+                Assert.True(await ke1);
+                Assert.True(await ku1);
+                Assert.False(await ke2);
+            }
+        }
+
+        [Fact]
+        public async Task DeleteMany()
+        {
+            using (var muxer = Create())
+            {
+                var db = muxer.GetDatabase();
+                var key1 = Me();
+                var key2 = Me() + "2";
+                var key3 = Me() + "3";
+                var ss = db.StringSetAsync(key1, "Heyyyyy");
+                var ss2 = db.StringSetAsync(key2, "Heyyyyy");
+                // key 3 not set
+                var ku1 = db.KeyDelete(new RedisKey[] { key1, key2, key3 });
+                var ke1 = db.KeyExistsAsync(key1).ForAwait();
+                var ke2 = db.KeyExistsAsync(key2).ForAwait();
+                Assert.Equal(2, ku1);
+                Assert.False(await ke1);
+                Assert.False(await ke2);
+            }
+        }
+
+        [Fact]
+        public async Task DeleteManyAsync()
+        {
+            using (var muxer = Create())
+            {
+                var db = muxer.GetDatabase();
+                var key1 = Me();
+                var key2 = Me() + "2";
+                var key3 = Me() + "3";
+                var ss = db.StringSetAsync(key1, "Heyyyyy");
+                var ss2 = db.StringSetAsync(key2, "Heyyyyy");
+                // key 3 not set
+                var ku1 = db.KeyDeleteAsync(new RedisKey[] { key1, key2, key3 }).ForAwait();
+                var ke1 = db.KeyExistsAsync(key1).ForAwait();
+                var ke2 = db.KeyExistsAsync(key2).ForAwait();
+                Assert.Equal(2, await ku1);
+                Assert.False(await ke1);
+                Assert.False(await ke2);
+            }
+        }
+
         [Fact]
         public void WrappedDatabasePrefixIntegration()
         {

StackExchange.Redis/StackExchange/Redis/Enums/RedisCommand.cs

@@ -161,6 +161,7 @@ internal enum RedisCommand
         TTL,
         TYPE,
 
+        UNLINK,
         UNSUBSCRIBE,
         UNWATCH,
 

StackExchange.Redis/StackExchange/Redis/Interfaces/IDatabase.cs

@@ -424,20 +424,24 @@ public interface IDatabase : IRedis, IDatabaseAsync
 
         /// <summary>
         /// Removes the specified key. A key is ignored if it does not exist.
+        /// If UNLINK is available (Redis 4.0+), it will be used.
         /// </summary>
         /// <param name="key">The key to delete.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>True if the key was removed.</returns>
         /// <remarks>https://redis.io/commands/del</remarks>
+        /// <remarks>https://redis.io/commands/unlink</remarks>
         bool KeyDelete(RedisKey key, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Removes the specified keys. A key is ignored if it does not exist.
+        /// If UNLINK is available (Redis 4.0+), it will be used.
         /// </summary>
         /// <param name="keys">The keys to delete.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>The number of keys that were removed.</returns>
         /// <remarks>https://redis.io/commands/del</remarks>
+        /// <remarks>https://redis.io/commands/unlink</remarks>
         long KeyDelete(RedisKey[] keys, CommandFlags flags = CommandFlags.None);
 
         /// <summary>

StackExchange.Redis/StackExchange/Redis/Interfaces/IDatabaseAsync.cs

@@ -388,20 +388,24 @@ public interface IDatabaseAsync : IRedisAsync
 
         /// <summary>
         /// Removes the specified key. A key is ignored if it does not exist.
+        /// If UNLINK is available (Redis 4.0+), it will be used.
         /// </summary>
         /// <param name="key">The key to delete.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>True if the key was removed.</returns>
         /// <remarks>https://redis.io/commands/del</remarks>
+        /// <remarks>https://redis.io/commands/unlink</remarks>
         Task<bool> KeyDeleteAsync(RedisKey key, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Removes the specified keys. A key is ignored if it does not exist.
+        /// If UNLINK is available (Redis 4.0+), it will be used.
         /// </summary>
         /// <param name="keys">The keys to delete.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>The number of keys that were removed.</returns>
         /// <remarks>https://redis.io/commands/del</remarks>
+        /// <remarks>https://redis.io/commands/unlink</remarks>
         Task<long> KeyDeleteAsync(RedisKey[] keys, CommandFlags flags = CommandFlags.None);
 
         /// <summary>

StackExchange.Redis/StackExchange/Redis/Message.cs

@@ -388,6 +388,7 @@ public static bool IsMasterOnly(RedisCommand command)
                 case RedisCommand.SREM:
                 case RedisCommand.SUNIONSTORE:
                 case RedisCommand.SWAPDB:
+                case RedisCommand.UNLINK:
                 case RedisCommand.ZADD:
                 case RedisCommand.ZINTERSTORE:
                 case RedisCommand.ZINCRBY:

StackExchange.Redis/StackExchange/Redis/RedisDatabase.cs

@@ -573,29 +573,50 @@ public bool IsConnected(RedisKey key, CommandFlags flags = CommandFlags.None)
 
         public bool KeyDelete(RedisKey key, CommandFlags flags = CommandFlags.None)
         {
-            var msg = Message.Create(Database, flags, RedisCommand.DEL, key);
-            return ExecuteSync(msg, ResultProcessor.DemandZeroOrOne);
+            var cmd = GetDeleteCommand(key, flags, out var server);
+            var msg = Message.Create(Database, flags, cmd, key);
+            return ExecuteSync(msg, ResultProcessor.DemandZeroOrOne, server);
         }
 
         public long KeyDelete(RedisKey[] keys, CommandFlags flags = CommandFlags.None)
         {
             if (keys == null) throw new ArgumentNullException(nameof(keys));
-            var msg = keys.Length == 0 ? null : Message.Create(Database, flags, RedisCommand.DEL, keys);
-            return ExecuteSync(msg, ResultProcessor.Int64);
+            if (keys.Length > 0)
+            {
+                var cmd = GetDeleteCommand(keys[0], flags, out var server);
+                var msg = keys.Length == 0 ? null : Message.Create(Database, flags, cmd, keys);
+                return ExecuteSync(msg, ResultProcessor.Int64, server);
+            }
+            return 0;
         }
 
         public Task<bool> KeyDeleteAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
         {
+            var cmd = GetDeleteCommand(key, flags, out var server);
             var msg = Message.Create(Database, flags, RedisCommand.DEL, key);
-            return ExecuteAsync(msg, ResultProcessor.DemandZeroOrOne);
+            return ExecuteAsync(msg, ResultProcessor.DemandZeroOrOne, server);
         }
 
         public Task<long> KeyDeleteAsync(RedisKey[] keys, CommandFlags flags = CommandFlags.None)
         {
             if (keys == null) throw new ArgumentNullException(nameof(keys));
+            if (keys.Length > 0)
+            {
+                var cmd = GetDeleteCommand(keys[0], flags, out var server);
+                var msg = keys.Length == 0 ? null : Message.Create(Database, flags, cmd, keys);
+                return ExecuteAsync(msg, ResultProcessor.Int64, server);
+            }
+            return CompletedTask<long>.Default(0);
+        }
 
-            var msg = keys.Length == 0 ? null : Message.Create(Database, flags, RedisCommand.DEL, keys);
-            return ExecuteAsync(msg, ResultProcessor.Int64);
+        private RedisCommand GetDeleteCommand(RedisKey key, CommandFlags flags, out ServerEndPoint server)
+        {
+            var features = GetFeatures(Database, key, flags, out server);
+            if (server != null && features.Unlink && multiplexer.CommandMap.IsAvailable(RedisCommand.UNLINK))
+            {
+                return RedisCommand.UNLINK;
+            }
+            return RedisCommand.DEL;
         }
 
         public byte[] KeyDump(RedisKey key, CommandFlags flags = CommandFlags.None)

StackExchange.Redis/StackExchange/Redis/RedisFeatures.cs

@@ -29,6 +29,7 @@ namespace StackExchange.Redis
                                          v2_9_5 = new Version(2, 9, 5),
                                          v3_0_0 = new Version(3, 0, 0),
                                          v3_2_0 = new Version(3, 2, 0),
+                                         v4_0_0 = new Version(4, 0, 0),
                                          v4_9_1 = new Version(4, 9, 1); // 5.0 RC1 is version 4.9.1
 
         private readonly Version version;
@@ -141,6 +142,11 @@ public RedisFeatures(Version version)
         /// </summary>
         public bool Time => Version >= v2_6_0;
 
+        /// <summary>
+        /// Does UNLINK exist?
+        /// </summary>
+        public bool Unlink => Version >= v4_0_0;
+
         /// <summary>
         /// Are Lua changes to the calling database transparent to the calling client?
         /// </summary>