more post-merge fixups

Author: mgravell

src/RESPite.StackExchange.Redis/RespContextServer.cs

@@ -159,6 +159,12 @@ public Task<RedisResult> ExecuteAsync(
         ICollection<object> args,
         CommandFlags flags = CommandFlags.None) => throw new NotImplementedException();
 
+    public RedisResult Execute(int? database, string command, ICollection<object> args, CommandFlags flags = CommandFlags.None) =>
+        throw new NotImplementedException();
+
+    public Task<RedisResult> ExecuteAsync(int? database, string command, ICollection<object> args, CommandFlags flags = CommandFlags.None)
+        => throw new NotImplementedException();
+
     public void FlushAllDatabases(CommandFlags flags = CommandFlags.None) => throw new NotImplementedException();
 
     public Task FlushAllDatabasesAsync(CommandFlags flags = CommandFlags.None) => throw new NotImplementedException();

src/RESPite.StackExchange.Redis/RespMultiplexer.cs

@@ -254,6 +254,16 @@ public IServer GetServer(EndPoint endpoint, object? asyncState = null)
         return GetServer(_connectionManager.GetNode(host, port), asyncState);
     }
 
+    public IServer GetServer(RedisKey key, object? asyncState = null, CommandFlags flags = CommandFlags.None)
+    {
+        if (key.IsNull) // just get anything
+        {
+            var node = _connectionManager.GetRandomNode();
+            if (node is not null) return GetServer(node, asyncState);
+        }
+        throw new NotImplementedException();
+    }
+
     private IServer GetServer(Node node, object? asyncState)
     {
         if (asyncState is not null) ThrowNotSupported();

src/RESPite/Connections/RespConnectionManager.cs

@@ -1,5 +1,6 @@
 using System.Buffers;
 using System.Diagnostics.CodeAnalysis;
+using System.Net;
 using RESPite.Connections.Internal;
 using RESPite.Internal;
 
@@ -99,6 +100,7 @@ internal Task ConnectAsync(RespConfiguration options, ReadOnlySpan<EndpointPair>
         {
             pending[i] = snapshot[i].ConnectAsync(log);
         }
+
         return ConnectAsyncAwaited(pending, log, snapshot.Length);
     }
 
@@ -230,4 +232,22 @@ internal Node GetNode(string host, int port)
     internal Node GetNode(string hostAndPort) => ConnectionFactory.TryParse(hostAndPort, out var host, out var port)
         ? GetNode(host, port)
         : throw new ArgumentException($"Could not parse host and port from '{hostAndPort}'", nameof(hostAndPort));
+
+    internal Node? GetRandomNode()
+    {
+        var nodes = _nodes;
+        if (nodes is { Length: > 0 })
+        {
+            var index = SharedRandom.Next(nodes.Length);
+            return nodes[index];
+        }
+
+        return null;
+    }
+
+#if NET5_0_OR_GREATER
+    private static Random SharedRandom => Random.Shared;
+#else
+    private static Random SharedRandom { get; } = new();
+#endif
 }

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -210,7 +210,7 @@ internal async Task MakePrimaryAsync(ServerEndPoint server, ReplicationChangeOpt
             {
                 throw ExceptionFactory.AdminModeNotEnabled(RawConfig.IncludeDetailInExceptions, cmd, null, server);
             }
-            var srv = new RedisServer(this, server, null);
+            var srv = server.GetRedisServer(null);
             if (!srv.IsConnected)
             {
                 throw ExceptionFactory.NoConnectionAvailable(this, null, server, GetServerSnapshot(), command: cmd);
@@ -1229,7 +1229,21 @@ public IServer GetServer(EndPoint? endpoint, object? asyncState = null)
                 throw new NotSupportedException($"The server API is not available via {RawConfig.Proxy}");
             }
             var server = servers[endpoint] as ServerEndPoint ?? throw new ArgumentException("The specified endpoint is not defined", nameof(endpoint));
-            return new RedisServer(this, server, asyncState);
+            return new RedisServer(server, asyncState);
+        }
+
+        /// <inheritdoc cref="IConnectionMultiplexer.GetServer(RedisKey, object, CommandFlags)"/>
+#pragma warning disable RS0026
+        public IServer GetServer(RedisKey key, object? asyncState = null, CommandFlags flags = CommandFlags.None)
+#pragma warning restore RS0026
+        {
+            // We'll spoof the GET command for this; we're not supporting ad-hoc access to the pub/sub channel, because: bad things.
+            // Any read-only-replica vs writable-primary concerns should be managed by the caller via "flags"; the default is PreferPrimary.
+            // Note that ServerSelectionStrategy treats "null" (default) keys as NoSlot, aka Any.
+            return (SelectServer(RedisCommand.GET, flags, key) ?? Throw()).GetRedisServer(asyncState);
+
+            [DoesNotReturn]
+            static ServerEndPoint Throw() => throw new InvalidOperationException("It was not possible to resolve a connection to the server owning the specified key");
         }
 
         /// <summary>
@@ -1241,7 +1255,7 @@ public IServer[] GetServers()
             var result = new IServer[snapshot.Length];
             for (var i = 0; i < snapshot.Length; i++)
             {
-                result[i] = new RedisServer(this, snapshot[i], null);
+                result[i] = snapshot[i].GetRedisServer(null);
             }
             return result;
         }

src/StackExchange.Redis/Interfaces/IConnectionMultiplexer.cs

@@ -213,6 +213,19 @@ public interface IConnectionMultiplexer : IDisposable, IAsyncDisposable
         /// <param name="asyncState">The async state to pass to the created <see cref="IServer"/>.</param>
         IServer GetServer(EndPoint endpoint, object? asyncState = null);
 
+        /// <summary>
+        /// Gets a server that would be used for a given key and flags.
+        /// </summary>
+        /// <param name="key">The endpoint to get a server for. In a non-cluster environment, this parameter is ignored. A <see langword="default"/> key may be specified
+        /// on cluster, which will return a connection to an arbitrary server matching the specified flags.</param>
+        /// <param name="asyncState">The async state to pass to the created <see cref="IServer"/>.</param>
+        /// <param name="flags">The command flags to use.</param>
+        /// <remarks>This method is particularly useful when communicating with a cluster environment, to obtain a connection to the server that owns the specified key
+        /// and ad-hoc commands with unusual routing requirements. Note that <see cref="GetDatabase"/> provides a connection that automatically routes commands by
+        /// looking for <see cref="RedisKey"/> parameters, so this method is only necessary when used with commands that do not take a <see cref="RedisKey"/> parameter,
+        /// but require consistent routing using key-like semantics.</remarks>
+        IServer GetServer(RedisKey key, object? asyncState = null, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Obtain configuration APIs for all servers in this multiplexer.
         /// </summary>

src/StackExchange.Redis/Interfaces/IServer.cs

@@ -252,10 +252,13 @@ public partial interface IServer : IRedis
         /// <inheritdoc cref="Echo(RedisValue, CommandFlags)"/>
         Task<RedisValue> EchoAsync(RedisValue message, CommandFlags flags = CommandFlags.None);
 
+#pragma warning disable RS0026, RS0027 // multiple overloads
         /// <summary>
         /// Execute an arbitrary command against the server; this is primarily intended for
         /// executing modules, but may also be used to provide access to new features that lack
-        /// a direct API.
+        /// a direct API. The command is assumed to be not database-specific. If this is not the case,
+        /// <see cref="Execute(int?, string, ICollection{object}, CommandFlags)"/> should be used to
+        /// specify the database (using <langword>null</langword> to use the configured default database).
         /// </summary>
         /// <param name="command">The command to run.</param>
         /// <param name="args">The arguments to pass for the command.</param>
@@ -280,6 +283,23 @@ public partial interface IServer : IRedis
 
         /// <inheritdoc cref="Execute(string, ICollection{object}, CommandFlags)"/>
         Task<RedisResult> ExecuteAsync(string command, ICollection<object> args, CommandFlags flags = CommandFlags.None);
+#pragma warning restore RS0026, RS0027
+
+        /// <summary>
+        /// Execute an arbitrary database-specific command against the server; this is primarily intended for
+        /// executing modules, but may also be used to provide access to new features that lack
+        /// a direct API.
+        /// </summary>
+        /// <param name="database">The database ID; if <see langword="null"/>, the configured default database is used.</param>
+        /// <param name="command">The command to run.</param>
+        /// <param name="args">The arguments to pass for the command.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>A dynamic representation of the command's result.</returns>
+        /// <remarks>This API should be considered an advanced feature; inappropriate use can be harmful.</remarks>
+        RedisResult Execute(int? database, string command, ICollection<object> args, CommandFlags flags = CommandFlags.None);
+
+        /// <inheritdoc cref="Execute(int?, string, ICollection{object}, CommandFlags)"/>
+        Task<RedisResult> ExecuteAsync(int? database, string command, ICollection<object> args, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Delete all the keys of all databases on the server.

src/StackExchange.Redis/RedisServer.cs

@@ -16,9 +16,9 @@ internal sealed class RedisServer : RedisBase, IServer
     {
         private readonly ServerEndPoint server;
 
-        internal RedisServer(ConnectionMultiplexer multiplexer, ServerEndPoint server, object? asyncState) : base(multiplexer, asyncState)
+        internal RedisServer(ServerEndPoint server, object? asyncState) : base(server.Multiplexer, asyncState)
         {
-            this.server = server ?? throw new ArgumentNullException(nameof(server));
+            this.server = server; // definitely can't be null because .Multiplexer in base call
         }
 
         int IServer.DatabaseCount => server.Databases;
@@ -1045,6 +1045,20 @@ public Task<RedisResult> ExecuteAsync(string command, ICollection<object> args,
             return ExecuteAsync(msg, ResultProcessor.ScriptResult, defaultValue: RedisResult.NullSingle);
         }
 
+        public RedisResult Execute(int? database, string command, ICollection<object> args, CommandFlags flags = CommandFlags.None)
+        {
+            var db = multiplexer.ApplyDefaultDatabase(database ?? -1);
+            var msg = new RedisDatabase.ExecuteMessage(multiplexer?.CommandMap, db, flags, command, args);
+            return ExecuteSync(msg, ResultProcessor.ScriptResult, defaultValue: RedisResult.NullSingle);
+        }
+
+        public Task<RedisResult> ExecuteAsync(int? database, string command, ICollection<object> args, CommandFlags flags = CommandFlags.None)
+        {
+            var db = multiplexer.ApplyDefaultDatabase(database ?? -1);
+            var msg = new RedisDatabase.ExecuteMessage(multiplexer?.CommandMap, db, flags, command, args);
+            return ExecuteAsync(msg, ResultProcessor.ScriptResult, defaultValue: RedisResult.NullSingle);
+        }
+
         /// <summary>
         /// For testing only.
         /// </summary>

src/StackExchange.Redis/ServerEndPoint.cs

@@ -71,6 +71,12 @@ public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint)
             }
         }
 
+        private RedisServer? _defaultServer;
+        public RedisServer GetRedisServer(object? asyncState)
+            => asyncState is null
+                ? (_defaultServer ??= new RedisServer(this, null)) // reuse and memoize
+                : new RedisServer(this, asyncState);
+
         public EndPoint EndPoint { get; }
 
         public ClusterConfiguration? ClusterConfiguration { get; private set; }

tests/BasicTest/RedisBenchmarks.cs

@@ -191,6 +191,7 @@ public void StringGet()
             db.StringGet(StringKey_K);
         }
     }
+
 #if !TEST_BASELINE
     /// <summary>
     /// Run StringSet lots of times.

tests/RESPite.Tests/ConnectionFixture.cs

@@ -163,6 +163,9 @@ IServer IConnectionMultiplexer.GetServer(string hostAndPort, object? asyncState)
     IServer IConnectionMultiplexer.GetServer(EndPoint endpoint, object? asyncState) =>
         throw new NotImplementedException();
 
+    public IServer GetServer(RedisKey key, object? asyncState = null, CommandFlags flags = CommandFlags.None)
+        => throw new NotImplementedException();
+
     IServer[] IConnectionMultiplexer.GetServers() => throw new NotImplementedException();
 
     Task<bool> IConnectionMultiplexer.ConfigureAsync(TextWriter? log) => throw new NotImplementedException();