Implement GetServer(RedisKey, ...) (#2936)

* - implement GetServer API from #2932, and tests
- implement memoization on GetServer when asyncState is null
- fixup NotImplementedException in dummy places

* release notes and shipped

* remove params overload

* remove params overload from muxer impl

* test stability: allow both embstr or raw for string encoding

* remove pragma

* Additional intellisense comments to clarify usage.

Author: mgravell

docs/ReleaseNotes.md

@@ -13,6 +13,7 @@ Current package versions:
 - Add `Condition.SortedSet[Not]ContainsStarting` condition for transactions ([#2638 by ArnoKoll](https://github.com/StackExchange/StackExchange.Redis/pull/2638))
 - Add support for XPENDING Idle time filter ([#2822 by david-brink-talogy](https://github.com/StackExchange/StackExchange.Redis/pull/2822))
 - Improve `double` formatting performance on net8+ ([#2928 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2928))
+- Add `GetServer(RedisKey, ...)` API ([#2936 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2936))
 - Fix error constructing `StreamAdd` message ([#2941 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2941)) 
 
 ## 2.8.58

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 server.GetRedisServer(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;
         }