Merge branch 'main' into marc/resp-reader

Author: mgravell

docs/Authentication.md

@@ -1,5 +1,4 @@
-Authentication
-===
+# Authentication
 
 There are multiple ways of connecting to a Redis server, depending on the authentication model. The simplest
 (but least secure) approach is to use the `default` user, with no authentication, and no transport security.
@@ -12,10 +11,9 @@ var muxer = await ConnectionMultiplexer.ConnectAsync("myserver"); // or myserver
 This approach is often used for local transient servers - it is simple, but insecure. But from there,
 we can get more complex!
 
-TLS
-===
+## TLS
 
-If your server has TLS enabled, SE.Redis can be instructed to use it. In some cases (AMR, etc), the
+If your server has TLS enabled, SE.Redis can be instructed to use it. In some cases (Azure Managed Redis, etc), the
 library will recognize the endpoint address, meaning: *you do not need to do anything*. To
 *manually* enable TLS, the `ssl` token can be used:
 
@@ -44,8 +42,7 @@ Alternatively, in advanced scenarios: to provide your own custom server validati
 can be used; this uses the normal [`RemoteCertificateValidationCallback`](https://learn.microsoft.com/dotnet/api/system.net.security.remotecertificatevalidationcallback)
 API.
 
-Usernames and Passwords
-===
+## Usernames and Passwords
 
 Usernames and passwords can be specified with the `user` and `password` tokens, respectively:
 
@@ -56,15 +53,25 @@ var muxer = await ConnectionMultiplexer.ConnectAsync("myserver,ssl=true,user=myu
 If no `user` is provided, the `default` user is assumed. In some cases, an authentication-token can be
 used in place of a classic password.
 
-Client certificates
-===
+## Managed identities
+
+If the server is an Azure Managed Redis resource, connections can be secured using Microsoft Entra ID authentication. Use the [Microsoft.Azure.StackExchangeRedis](https://github.com/Azure/Microsoft.Azure.StackExchangeRedis) extension package to handle the authentication using tokens retrieved from Microsoft Entra. The package integrates via the ConfigurationOptions class, and can use various types of identities for token retrieval. For example with a user-assigned managed identity: 
+
+```csharp
+var options = ConfigurationOptions.Parse("mycache.region.redis.azure.net:10000");
+await options.ConfigureForAzureWithUserAssignedManagedIdentityAsync(managedIdentityClientId);
+```
+
+For details and samples see [https://github.com/Azure/Microsoft.Azure.StackExchangeRedis](https://github.com/Azure/Microsoft.Azure.StackExchangeRedis)
+
+## Client certificates
 
 If the server is configured to require a client certificate, this can be supplied in multiple ways.
 If you have a local public / private key pair (such as `MyUser2.crt` and `MyUser2.key`), the
 `options.SetUserPemCertificate(...)` method can be used:
 
 ``` csharp
-config.SetUserPemCertificate(
+options.SetUserPemCertificate(
     userCertificatePath: userCrtPath,
     userKeyPath: userKeyPath
 );
@@ -74,7 +81,7 @@ If you have a single `pfx` file that contains the public / private pair, the `op
 method can be used:
 
 ``` csharp
-config.SetUserPfxCertificate(
+options.SetUserPfxCertificate(
     userCertificatePath: userCrtPath,
     password: filePassword // optional
 );
@@ -85,8 +92,7 @@ can be used; this uses the normal
 [`LocalCertificateSelectionCallback`](https://learn.microsoft.com/dotnet/api/system.net.security.remotecertificatevalidationcallback)
 API.
 
-User certificates with implicit user authentication
-===
+## User certificates with implicit user authentication
 
 Historically, the client certificate only provided access to the server, but as the `default` user. From 8.6,
 the server can be configured to use client certificates to provide user identity. This replaces the
@@ -114,8 +120,7 @@ var user = (string?)await conn.GetDatabase().ExecuteAsync("acl", "whoami");
 Console.WriteLine(user); // writes "MyUser2"
 ```
 
-More info
-===
+## More info
 
 For more information:
 

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -101,6 +101,22 @@ protected override void OnMoved(RedisClient client, int hashSlot, Node node)
         base.OnMoved(client, hashSlot, node);
     }
 
+    protected override void OnOutOfBand(RedisClient client, TypedRedisValue message)
+    {
+        if (message.IsAggregate
+            && message.Span is { IsEmpty: false } span
+            && !span[0].IsAggregate)
+        {
+            _log?.WriteLine($"Client {client.Id}: {span[0].AsRedisValue()} {message} ");
+        }
+        else
+        {
+            _log?.WriteLine($"Client {client.Id}: {message}");
+        }
+
+        base.OnOutOfBand(client, message);
+    }
+
     public override TypedRedisValue OnUnknownCommand(in RedisClient client, in RedisRequest request, ReadOnlySpan<byte> command)
     {
         _log?.WriteLine($"[{client.Id}] unknown command: {Encoding.ASCII.GetString(command)}");

tests/StackExchange.Redis.Tests/PubSubTests.cs

@@ -11,12 +11,31 @@
 namespace StackExchange.Redis.Tests;
 
 [RunPerProtocol]
-public class PubSubTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture)
+public class PubSubTests(ITestOutputHelper output, SharedConnectionFixture fixture)
+    : PubSubTestBase(output, fixture, null)
+{
+}
+
+/*
+[RunPerProtocol]
+public class InProcPubSubTests(ITestOutputHelper output, InProcServerFixture fixture)
+    : PubSubTestBase(output, null, fixture)
+{
+    protected override bool UseDedicatedInProcessServer => false;
+}
+*/
+
+[RunPerProtocol]
+public abstract class PubSubTestBase(
+    ITestOutputHelper output,
+    SharedConnectionFixture? connection,
+    InProcServerFixture? server)
+    : TestBase(output, connection, server)
 {
     [Fact]
     public async Task ExplicitPublishMode()
     {
-        await using var conn = Create(channelPrefix: "foo:", log: Writer);
+        await using var conn = ConnectFactory(channelPrefix: "foo:");
 
         var pub = conn.GetSubscriber();
         int a = 0, b = 0, c = 0, d = 0;
@@ -54,9 +73,9 @@ await UntilConditionAsync(
     [InlineData("Foo:", true, "f")]
     public async Task TestBasicPubSub(string? channelPrefix, bool wildCard, string breaker)
     {
-        await using var conn = Create(channelPrefix: channelPrefix, shared: false, log: Writer);
+        await using var conn = ConnectFactory(channelPrefix: channelPrefix, shared: false);
 
-        var pub = GetAnyPrimary(conn);
+        var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
         await PingAsync(pub, sub).ForAwait();
         HashSet<string?> received = [];
@@ -139,10 +158,10 @@ public async Task TestBasicPubSub(string? channelPrefix, bool wildCard, string b
     [Fact]
     public async Task TestBasicPubSubFireAndForget()
     {
-        await using var conn = Create(shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
-        var profiler = conn.AddProfiler();
-        var pub = GetAnyPrimary(conn);
+        var profiler = conn.DefaultClient.AddProfiler();
+        var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
 
         RedisChannel key = RedisChannel.Literal(Me() + Guid.NewGuid());
@@ -214,9 +233,9 @@ private async Task PingAsync(IServer pub, ISubscriber sub, int times = 1)
     [Fact]
     public async Task TestPatternPubSub()
     {
-        await using var conn = Create(shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
-        var pub = GetAnyPrimary(conn);
+        var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
 
         HashSet<string?> received = [];
@@ -273,7 +292,7 @@ public async Task TestPatternPubSub()
     [Fact]
     public async Task TestPublishWithNoSubscribers()
     {
-        await using var conn = Create();
+        await using var conn = ConnectFactory();
 
         var sub = conn.GetSubscriber();
 #pragma warning disable CS0618
@@ -285,7 +304,7 @@ public async Task TestPublishWithNoSubscribers()
     public async Task TestMassivePublishWithWithoutFlush_Local()
     {
         Skip.UnlessLongRunning();
-        await using var conn = Create();
+        await using var conn = ConnectFactory();
 
         var sub = conn.GetSubscriber();
         TestMassivePublish(sub, Me(), "local");
@@ -335,7 +354,7 @@ private void TestMassivePublish(ISubscriber sub, string channel, string caption)
     [Fact]
     public async Task SubscribeAsyncEnumerable()
     {
-        await using var conn = Create(syncTimeout: 20000, shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
         var sub = conn.GetSubscriber();
         RedisChannel channel = RedisChannel.Literal(Me());
@@ -370,7 +389,7 @@ public async Task SubscribeAsyncEnumerable()
     [Fact]
     public async Task PubSubGetAllAnyOrder()
     {
-        await using var conn = Create(syncTimeout: 20000, shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
         var sub = conn.GetSubscriber();
         RedisChannel channel = RedisChannel.Literal(Me());
@@ -625,9 +644,10 @@ public async Task PubSubGetAllCorrectOrder_OnMessage_Async()
     [Fact]
     public async Task TestPublishWithSubscribers()
     {
-        await using var connA = Create(shared: false, log: Writer);
-        await using var connB = Create(shared: false, log: Writer);
-        await using var connPub = Create();
+        await using var pair = ConnectFactory(shared: false);
+        await using var connA = pair.DefaultClient;
+        await using var connB = pair.CreateClient();
+        await using var connPub = pair.CreateClient();
 
         var channel = Me();
         var listenA = connA.GetSubscriber();
@@ -652,9 +672,10 @@ public async Task TestPublishWithSubscribers()
     [Fact]
     public async Task TestMultipleSubscribersGetMessage()
     {
-        await using var connA = Create(shared: false, log: Writer);
-        await using var connB = Create(shared: false, log: Writer);
-        await using var connPub = Create();
+        await using var pair = ConnectFactory(shared: false);
+        await using var connA = pair.DefaultClient;
+        await using var connB = pair.CreateClient();
+        await using var connPub = pair.CreateClient();
 
         var channel = RedisChannel.Literal(Me());
         var listenA = connA.GetSubscriber();
@@ -682,7 +703,7 @@ public async Task TestMultipleSubscribersGetMessage()
     [Fact]
     public async Task Issue38()
     {
-        await using var conn = Create(log: Writer);
+        await using var conn = ConnectFactory();
 
         var sub = conn.GetSubscriber();
         int count = 0;
@@ -717,9 +738,10 @@ public async Task Issue38()
     [Fact]
     public async Task TestPartialSubscriberGetMessage()
     {
-        await using var connA = Create();
-        await using var connB = Create();
-        await using var connPub = Create();
+        await using var pair = ConnectFactory();
+        await using var connA = pair.DefaultClient;
+        await using var connB = pair.CreateClient();
+        await using var connPub = pair.CreateClient();
 
         int gotA = 0, gotB = 0;
         var listenA = connA.GetSubscriber();
@@ -750,8 +772,9 @@ public async Task TestPartialSubscriberGetMessage()
     [Fact]
     public async Task TestSubscribeUnsubscribeAndSubscribeAgain()
     {
-        await using var connPub = Create();
-        await using var connSub = Create();
+        await using var pair = ConnectFactory();
+        await using var connPub = pair.DefaultClient;
+        await using var connSub = pair.CreateClient();
 
         var prefix = Me();
         var pub = connPub.GetSubscriber();

tests/StackExchange.Redis.Tests/TestBase.cs

@@ -583,4 +583,74 @@ protected static async Task UntilConditionAsync(TimeSpan maxWaitTime, Func<bool>
             spent += wait;
         }
     }
+
+    // simplified usage to get an interchangeable dedicated vs shared in-process server, useful for debugging
+    protected virtual bool UseDedicatedInProcessServer => false; // use the shared server by default
+    internal ClientFactory ConnectFactory(bool allowAdmin = false, string? channelPrefix = null, bool shared = true)
+    {
+        if (UseDedicatedInProcessServer)
+        {
+            var server = new InProcessTestServer(Output);
+            return new ClientFactory(this, allowAdmin, channelPrefix, shared, server);
+        }
+        return new ClientFactory(this, allowAdmin, channelPrefix, shared, null);
+    }
+
+    internal sealed class ClientFactory : IDisposable, IAsyncDisposable
+    {
+        private readonly TestBase _testBase;
+        private readonly bool _allowAdmin;
+        private readonly string? _channelPrefix;
+        private readonly bool _shared;
+        private readonly InProcessTestServer? _server;
+        private IInternalConnectionMultiplexer? _defaultClient;
+
+        internal ClientFactory(TestBase testBase, bool allowAdmin, string? channelPrefix, bool shared, InProcessTestServer? server)
+        {
+            _testBase = testBase;
+            _allowAdmin = allowAdmin;
+            _channelPrefix = channelPrefix;
+            _shared = shared;
+            _server = server;
+        }
+
+        public IInternalConnectionMultiplexer DefaultClient => _defaultClient ??= CreateClient();
+
+        public InProcessTestServer? Server => _server;
+
+        public IInternalConnectionMultiplexer CreateClient()
+        {
+            if (_server is not null)
+            {
+                var config = _server.GetClientConfig();
+                config.AllowAdmin = _allowAdmin;
+                if (_channelPrefix is not null)
+                {
+                    config.ChannelPrefix = RedisChannel.Literal(_channelPrefix);
+                }
+                return ConnectionMultiplexer.ConnectAsync(config).Result;
+            }
+            return _testBase.Create(allowAdmin: _allowAdmin, channelPrefix: _channelPrefix, shared: _shared);
+        }
+
+        public IDatabase GetDatabase(int db = -1) => DefaultClient.GetDatabase(db);
+
+        public ISubscriber GetSubscriber() => DefaultClient.GetSubscriber();
+
+        public void Dispose()
+        {
+            _server?.Dispose();
+            _defaultClient?.Dispose();
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            _server?.Dispose();
+            if (_defaultClient is not null)
+            {
+                return _defaultClient.DisposeAsync();
+            }
+            return default;
+        }
+    }
 }