Support Lazy Connect (#104)

Signed-off-by: currantw <taylor.curran@improving.com>

Author: currantw

DEVELOPER.md

@@ -212,7 +212,7 @@ cargo fmt --all -- --check
 
 6. Test framework and style
 
-The CSharp Valkey-Glide client uses xUnit v3 for testing code. The test code styles are defined in `.editorcofing` (see `dotnet_diagnostic.xUnit..` rules). The xUnit rules are enforced by the [xUnit analyzers](https://github.com/xunit/xunit.analyzers) referenced in the main xunit.v3 NuGet package. If you choose to use xunit.v3.core instead, you can reference xunit.analyzers explicitly. For additional info, please, refer to https://xunit.net and https://github.com/xunit/xunit
+The CSharp Valkey-Glide client uses xUnit v3 for testing code. The test code styles are defined in `.editorconfig` (see `dotnet_diagnostic.xUnit..` rules). The xUnit rules are enforced by the [xUnit analyzers](https://github.com/xunit/xunit.analyzers) referenced in the main xunit.v3 NuGet package. If you choose to use xunit.v3.core instead, you can reference xunit.analyzers explicitly. For additional info, please, refer to <https://xunit.net> and <https://github.com/xunit/xunit>
 
 ## Community and Feedback
 

rust/src/ffi.rs

@@ -71,6 +71,7 @@ pub struct ConnectionConfig {
     pub protocol: redis::ProtocolVersion,
     /// zero pointer is valid, means no client name is given (`None`)
     pub client_name: *const c_char,
+    pub lazy_connect: bool,
     /*
     TODO below
     pub periodic_checks: Option<PeriodicCheck>,
@@ -147,11 +148,11 @@ pub(crate) unsafe fn create_connection_request(
         } else {
             None
         },
+        lazy_connect: config.lazy_connect,
         // TODO below
         periodic_checks: None,
         pubsub_subscriptions: None,
         inflight_requests_limit: None,
-        lazy_connect: false,
     }
 }
 

sources/Valkey.Glide/BaseClient.GenericCommands.cs

@@ -178,7 +178,7 @@ public async Task<bool> KeyCopyAsync(ValkeyKey sourceKey, ValkeyKey destinationK
     public async Task<ValkeyValue[]> SortAsync(ValkeyKey key, long skip = 0, long take = -1, Order order = Order.Ascending, SortType sortType = SortType.Numeric, ValkeyValue by = default, ValkeyValue[]? get = null, CommandFlags flags = CommandFlags.None)
     {
         Utils.Requires<NotImplementedException>(flags == CommandFlags.None, "Command flags are not supported by GLIDE");
-        return await Command(Request.SortAsync(key, skip, take, order, sortType, by, get, _serverVersion));
+        return await Command(Request.SortAsync(key, skip, take, order, sortType, by, get, await GetServerVersionAsync()));
     }
 
     public async Task<long> SortAndStoreAsync(ValkeyKey destination, ValkeyKey key, long skip = 0, long take = -1, Order order = Order.Ascending, SortType sortType = SortType.Numeric, ValkeyValue by = default, ValkeyValue[]? get = null, CommandFlags flags = CommandFlags.None)

sources/Valkey.Glide/BaseClient.cs

@@ -53,14 +53,12 @@ protected static async Task<T> CreateClient<T>(BaseClientConfiguration config, F
         CreateClientFfi(request.ToPtr(), successCallbackPointer, failureCallbackPointer);
         client._clientPointer = await message; // This will throw an error thru failure callback if any
 
-        if (client._clientPointer != IntPtr.Zero)
+        if (client._clientPointer == IntPtr.Zero)
         {
-            // Initialize server version after successful connection
-            await client.InitializeServerVersionAsync();
-            return client;
+            throw new ConnectionException("Failed creating a client");
         }
 
-        throw new ConnectionException("Failed creating a client");
+        return client;
     }
 
     protected BaseClient()
@@ -127,6 +125,11 @@ internal virtual async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = n
 
         // All memory allocated is auto-freed by `using` operator
     }
+    protected Version? ParseServerVersion(string response)
+    {
+        var versionMatch = System.Text.RegularExpressions.Regex.Match(response, @"(?:valkey_version|redis_version):([\d\.]+)");
+        return versionMatch.Success ? new(versionMatch.Groups[1].Value) : null;
+    }
     #endregion protected methods
 
     #region private methods
@@ -145,7 +148,7 @@ private void FailureCallback(ulong index, IntPtr strPtr, RequestErrorType errTyp
 
     internal void SetInfo(string info) => _clientInfo = info;
 
-    protected abstract Task InitializeServerVersionAsync();
+    protected abstract Task<Version> GetServerVersionAsync();
 
     [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
     private delegate void SuccessAction(ulong index, IntPtr ptr);
@@ -169,6 +172,7 @@ private void FailureCallback(ulong index, IntPtr strPtr, RequestErrorType errTyp
     private readonly object _lock = new();
     private string _clientInfo = ""; // used to distinguish and identify clients during tests
     protected Version? _serverVersion; // cached server version
+    protected static readonly Version DefaultServerVersion = new(8, 0, 0);
 
     #endregion private fields
 }

sources/Valkey.Glide/ConnectionConfiguration.cs

@@ -24,9 +24,10 @@ internal record ConnectionConfig
         public uint DatabaseId;
         public Protocol? Protocol;
         public string? ClientName;
+        public bool LazyConnect;
 
         internal FFI.ConnectionConfig ToFfi() =>
-            new(Addresses, TlsMode, ClusterMode, (uint?)RequestTimeout?.TotalMilliseconds, (uint?)ConnectionTimeout?.TotalMilliseconds, ReadFrom, RetryStrategy, AuthenticationInfo, DatabaseId, Protocol, ClientName);
+            new(Addresses, TlsMode, ClusterMode, (uint?)RequestTimeout?.TotalMilliseconds, (uint?)ConnectionTimeout?.TotalMilliseconds, ReadFrom, RetryStrategy, AuthenticationInfo, DatabaseId, Protocol, ClientName, LazyConnect);
     }
 
     /// <summary>
@@ -171,7 +172,7 @@ public abstract class BaseClientConfiguration
     /// <summary>
     /// Configuration for a standalone client. <br />
     /// Use <see cref="StandaloneClientConfigurationBuilder" /> or
-    /// <see cref="StandaloneClientConfiguration(List{ValueTuple{string?, ushort?}}, bool?, TimeSpan?, TimeSpan?, ReadFrom?, RetryStrategy?, string?, string?, uint?, Protocol?, string?)" /> to create an instance.
+    /// <see cref="StandaloneClientConfiguration(List{ValueTuple{string?, ushort?}}, bool?, TimeSpan?, TimeSpan?, ReadFrom?, RetryStrategy?, string?, string?, uint?, Protocol?, string?, bool)" /> to create an instance.
     /// </summary>
     public sealed class StandaloneClientConfiguration : BaseClientConfiguration
     {
@@ -190,6 +191,7 @@ internal StandaloneClientConfiguration() { }
         /// <param name="databaseId"><inheritdoc cref="ClientConfigurationBuilder{T}.DataBaseId" path="/summary" /></param>
         /// <param name="protocol"><inheritdoc cref="ClientConfigurationBuilder{T}.ProtocolVersion" path="/summary" /></param>
         /// <param name="clientName"><inheritdoc cref="ClientConfigurationBuilder{T}.ClientName" path="/summary" /></param>
+        /// <param name="lazyConnect"><inheritdoc cref="ClientConfigurationBuilder{T}.LazyConnect" path="/summary" /></param>
         public StandaloneClientConfiguration(
             List<(string? host, ushort? port)> addresses,
             bool? useTls = null,
@@ -201,7 +203,8 @@ public StandaloneClientConfiguration(
             string? password = null,
             uint? databaseId = null,
             Protocol? protocol = null,
-            string? clientName = null
+            string? clientName = null,
+            bool lazyConnect = false
             )
         {
             StandaloneClientConfigurationBuilder builder = new();
@@ -215,6 +218,7 @@ public StandaloneClientConfiguration(
             _ = databaseId.HasValue ? builder.DataBaseId = databaseId.Value : new();
             _ = protocol.HasValue ? builder.ProtocolVersion = protocol.Value : new();
             _ = clientName is not null ? builder.ClientName = clientName : "";
+            builder.LazyConnect = lazyConnect;
             Request = builder.Build().Request;
         }
     }
@@ -239,6 +243,7 @@ internal ClusterClientConfiguration() { }
         /// <param name="databaseId"><inheritdoc cref="ClientConfigurationBuilder{T}.DataBaseId" path="/summary" /></param>
         /// <param name="protocol"><inheritdoc cref="ClientConfigurationBuilder{T}.ProtocolVersion" path="/summary" /></param>
         /// <param name="clientName"><inheritdoc cref="ClientConfigurationBuilder{T}.ClientName" path="/summary" /></param>
+        /// <param name="lazyConnect"><inheritdoc cref="ClientConfigurationBuilder{T}.LazyConnect" path="/summary" /></param>
         public ClusterClientConfiguration(
             List<(string? host, ushort? port)> addresses,
             bool? useTls = null,
@@ -250,7 +255,8 @@ public ClusterClientConfiguration(
             string? password = null,
             uint? databaseId = null,
             Protocol? protocol = null,
-            string? clientName = null
+            string? clientName = null,
+            bool lazyConnect = false
             )
         {
             ClusterClientConfigurationBuilder builder = new();
@@ -264,6 +270,7 @@ public ClusterClientConfiguration(
             _ = databaseId.HasValue ? builder.DataBaseId = databaseId.Value : new();
             _ = protocol.HasValue ? builder.ProtocolVersion = protocol.Value : new();
             _ = clientName is not null ? builder.ClientName = clientName : "";
+            builder.LazyConnect = lazyConnect;
             Request = builder.Build().Request;
         }
     }
@@ -534,6 +541,24 @@ public T WithDataBaseId(uint dataBaseId)
             return (T)this;
         }
         #endregion
+        #region Lazy Connect
+        /// <summary>
+        /// Configure whether to defer connections until the first command is executed.<br />
+        /// If not explicitly set, a default value of <c>false</c> will be used.
+        /// </summary>
+        public bool LazyConnect
+        {
+            get => Config.LazyConnect;
+            set => Config.LazyConnect = value;
+        }
+
+        /// <inheritdoc cref="LazyConnect" />
+        public T WithLazyConnect(bool lazyConnect)
+        {
+            LazyConnect = lazyConnect;
+            return (T)this;
+        }
+        #endregion
 
         internal ConnectionConfig Build() => Config;
     }

sources/Valkey.Glide/GlideClient.cs

@@ -196,21 +196,21 @@ public async IAsyncEnumerable<ValkeyKey> KeysAsync(int database = -1, ValkeyValu
         } while (currentCursor != 0);
     }
 
-    protected override async Task InitializeServerVersionAsync()
+    protected override async Task<Version> GetServerVersionAsync()
     {
-        try
+        if (_serverVersion == null)
         {
-            var infoResponse = await Command(Request.Info([InfoOptions.Section.SERVER]));
-            var versionMatch = System.Text.RegularExpressions.Regex.Match(infoResponse, @"(?:valkey_version|redis_version):([\d\.]+)");
-            if (versionMatch.Success)
+            try
             {
-                _serverVersion = new Version(versionMatch.Groups[1].Value);
+                var infoResponse = await Command(Request.Info([InfoOptions.Section.SERVER]));
+                _serverVersion = ParseServerVersion(infoResponse) ?? DefaultServerVersion;
+            }
+            catch
+            {
+                _serverVersion = DefaultServerVersion;
             }
         }
-        catch
-        {
-            // If we can't get version, assume newer version (use SORT_RO)
-            _serverVersion = new Version(8, 0, 0);
-        }
+
+        return _serverVersion;
     }
 }

sources/Valkey.Glide/GlideClusterClient.cs

@@ -301,21 +301,21 @@ public async Task<string> SelectAsync(long index, CommandFlags flags = CommandFl
         return await Command(Request.Select(index), Route.Random);
     }
 
-    protected override async Task InitializeServerVersionAsync()
+    protected override async Task<Version> GetServerVersionAsync()
     {
-        try
+        if (_serverVersion == null)
         {
-            var infoResponse = await Command(Request.Info([InfoOptions.Section.SERVER]).ToClusterValue(true), Route.Random);
-            var versionMatch = System.Text.RegularExpressions.Regex.Match(infoResponse.SingleValue, @"(?:valkey_version|redis_version):([\d\.]+)");
-            if (versionMatch.Success)
+            try
             {
-                _serverVersion = new Version(versionMatch.Groups[1].Value);
+                var infoResponse = await Command(Request.Info([InfoOptions.Section.SERVER]).ToClusterValue(true), Route.Random);
+                _serverVersion = ParseServerVersion(infoResponse.SingleValue) ?? DefaultServerVersion;
+            }
+            catch
+            {
+                _serverVersion = DefaultServerVersion;
             }
         }
-        catch
-        {
-            // If we can't get version, assume newer version (use SORT_RO)
-            _serverVersion = new Version(8, 0, 0);
-        }
+
+        return _serverVersion;
     }
 }

sources/Valkey.Glide/Internals/FFI.structs.cs

@@ -212,7 +212,8 @@ public ConnectionConfig(
             AuthenticationInfo? authenticationInfo,
             uint databaseId,
             ConnectionConfiguration.Protocol? protocol,
-            string? clientName)
+            string? clientName,
+            bool lazyConnect = false)
         {
             _addresses = addresses;
             _request = new()
@@ -235,6 +236,7 @@ public ConnectionConfig(
                 HasProtocol = protocol.HasValue,
                 Protocol = protocol ?? default,
                 ClientName = clientName,
+                LazyConnect = lazyConnect,
             };
         }
 
@@ -770,6 +772,8 @@ private struct ConnectionRequest
         public ConnectionConfiguration.Protocol Protocol;
         [MarshalAs(UnmanagedType.LPStr)]
         public string? ClientName;
+        [MarshalAs(UnmanagedType.U1)]
+        public bool LazyConnect;
         // TODO more config params, see ffi.rs
     }
 

tests/Valkey.Glide.IntegrationTests/ClusterClientTests.cs

@@ -5,6 +5,7 @@
 using Valkey.Glide.Pipeline;
 
 using static Valkey.Glide.Commands.Options.InfoOptions;
+using static Valkey.Glide.ConnectionConfiguration;
 using static Valkey.Glide.Errors;
 using static Valkey.Glide.Route;
 
@@ -585,4 +586,70 @@ public async Task TestKeyCopyAsync(GlideClusterClient client)
         // Verify the source key still exists in the current database
         Assert.True(await client.KeyExistsAsync(sourceKey));
     }
+
+    [Fact]
+    public async Task LazyConnect()
+    {
+        string serverName = $"test_{Guid.NewGuid():N}";
+
+        try
+        {
+            // Create dedicated server.
+            var addresses = ServerManager.StartClusterServer(serverName);
+            var configBuilder = new ClusterClientConfigurationBuilder()
+                .WithAddress(addresses[0].host, addresses[0].port);
+            var eagerConfig = configBuilder.WithLazyConnect(false).Build();
+            var lazyConfig = configBuilder.WithLazyConnect(true).Build();
+
+            // Create reference client.
+            using var referenceClient = await GlideClusterClient.CreateClient(eagerConfig);
+            var initialCount = await ConnectionInfo.GetConnectionCount(referenceClient);
+
+            // Create lazy client (does not connect immediately).
+            using var lazyClient = await GlideClusterClient.CreateClient(lazyConfig);
+            var connectCount = await ConnectionInfo.GetConnectionCount(referenceClient);
+
+            Assert.Equal(initialCount, connectCount);
+
+            // First command establishes connection.
+            await lazyClient.PingAsync();
+            var commandCount = await ConnectionInfo.GetConnectionCount(referenceClient);
+
+            Assert.True(connectCount < commandCount);
+        }
+        finally
+        {
+            ServerManager.StopServer(serverName);
+        }
+    }
+
+    [Fact]
+    public async Task EagerConnect()
+    {
+        string serverName = $"test_{Guid.NewGuid():N}";
+
+        try
+        {
+            // Create dedicated server.
+            var addresses = ServerManager.StartClusterServer(serverName);
+            var eagerConfig = new ClusterClientConfigurationBuilder()
+                .WithAddress(addresses[0].host, addresses[0].port)
+                .WithLazyConnect(false)
+                .Build();
+
+            // Create reference client.
+            using var referenceClient = await GlideClusterClient.CreateClient(eagerConfig);
+            var initialCount = await ConnectionInfo.GetConnectionCount(referenceClient);
+
+            // Create eager client (connects immediately).
+            using var eagerClient = await GlideClusterClient.CreateClient(eagerConfig);
+            var connectCount = await ConnectionInfo.GetConnectionCount(referenceClient);
+
+            Assert.True(initialCount < connectCount);
+        }
+        finally
+        {
+            ServerManager.StopServer(serverName);
+        }
+    }
 }

tests/Valkey.Glide.IntegrationTests/ConnectionInfo.cs

@@ -0,0 +1,34 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+namespace Valkey.Glide.IntegrationTests;
+
+/// <summary>
+/// Utility class for retrieving connection information from Valkey servers.
+/// </summary>
+internal static class ConnectionInfo
+{
+    private static readonly GlideString[] ClientListCommandArgs = ["CLIENT", "LIST"];
+
+    /// <summary>
+    /// Returns the total number of client connections to a standalone Valkey server.
+    /// </summary>
+    /// <param name="client">The standalone client instance.</param>
+    /// <returns>A task that resolves to the total number of client connections.</returns>
+    public static async Task<int> GetConnectionCount(GlideClient client)
+    {
+        var result = await client.CustomCommand(ClientListCommandArgs);
+        return result!.ToString()!.Split('\n', StringSplitOptions.RemoveEmptyEntries).Length;
+    }
+
+    /// <summary>
+    /// Returns the total number of client connections to a cluster Valkey server.
+    /// </summary>
+    /// <param name="client">The cluster client instance.</param>
+    /// <returns>A task that resolves to the total number of client connections.</returns>
+    public static async Task<int> GetConnectionCount(GlideClusterClient client)
+    {
+        var result = await client.CustomCommand(ClientListCommandArgs, new Route.AllPrimariesRoute());
+        return result!.MultiValue.Values.Sum(nodeResult =>
+            nodeResult!.ToString()!.Split('\n', StringSplitOptions.RemoveEmptyEntries).Length);
+    }
+}