Make ClientPointer and MessageContainer internal (including rename to PascalCase), move ClusterScanCommand to GlideClusterClient.

Signed-off-by: currantw <[email protected]>

Author: currantw

sources/Valkey.Glide/BaseClient.cs

@@ -22,21 +22,21 @@ public void Dispose()
         GC.SuppressFinalize(this);
         lock (_lock)
         {
-            if (_clientPointer == IntPtr.Zero)
+            if (ClientPointer == IntPtr.Zero)
             {
                 return;
             }
-            _messageContainer.DisposeWithError(null);
-            CloseClientFfi(_clientPointer);
-            _clientPointer = IntPtr.Zero;
+            MessageContainer.DisposeWithError(null);
+            CloseClientFfi(ClientPointer);
+            ClientPointer = IntPtr.Zero;
         }
     }
 
     public async ValueTask DisposeAsync() => await Task.Run(Dispose);
 
-    public override string ToString() => $"{GetType().Name} {{ 0x{_clientPointer:X} {_clientInfo} }}";
+    public override string ToString() => $"{GetType().Name} {{ 0x{ClientPointer:X} {_clientInfo} }}";
 
-    public override int GetHashCode() => (int)_clientPointer;
+    public override int GetHashCode() => (int)ClientPointer;
 
     #endregion public methods
 
@@ -49,11 +49,11 @@ protected static async Task<T> CreateClient<T>(BaseClientConfiguration config, F
         nint failureCallbackPointer = Marshal.GetFunctionPointerForDelegate(client._failureCallbackDelegate);
 
         using FFI.ConnectionConfig request = config.Request.ToFfi();
-        Message message = client._messageContainer.GetMessageForCall();
+        Message message = client.MessageContainer.GetMessageForCall();
         CreateClientFfi(request.ToPtr(), successCallbackPointer, failureCallbackPointer);
-        client._clientPointer = await message; // This will throw an error thru failure callback if any
+        client.ClientPointer = await message; // This will throw an error thru failure callback if any
 
-        if (client._clientPointer == IntPtr.Zero)
+        if (client.ClientPointer == IntPtr.Zero)
         {
             throw new ConnectionException("Failed creating a client");
         }
@@ -65,7 +65,7 @@ protected BaseClient()
     {
         _successCallbackDelegate = SuccessCallback;
         _failureCallbackDelegate = FailureCallback;
-        _messageContainer = new(this);
+        MessageContainer = new(this);
     }
 
     protected internal delegate T ResponseHandler<T>(IntPtr response);
@@ -83,8 +83,8 @@ internal virtual async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = n
         using FFI.Route? ffiRoute = route?.ToFfi();
 
         // 3. Sumbit request to the rust part
-        Message message = _messageContainer.GetMessageForCall();
-        CommandFfi(_clientPointer, (ulong)message.Index, cmd.ToPtr(), ffiRoute?.ToPtr() ?? IntPtr.Zero);
+        Message message = MessageContainer.GetMessageForCall();
+        CommandFfi(ClientPointer, (ulong)message.Index, cmd.ToPtr(), ffiRoute?.ToPtr() ?? IntPtr.Zero);
 
         // 4. Get a response and Handle it
         IntPtr response = await message;
@@ -109,8 +109,8 @@ internal virtual async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = n
         using FFI.BatchOptions? ffiOptions = options?.ToFfi();
 
         // 3. Sumbit request to the rust part
-        Message message = _messageContainer.GetMessageForCall();
-        BatchFfi(_clientPointer, (ulong)message.Index, ffiBatch.ToPtr(), raiseOnError, ffiOptions?.ToPtr() ?? IntPtr.Zero);
+        Message message = MessageContainer.GetMessageForCall();
+        BatchFfi(ClientPointer, (ulong)message.Index, ffiBatch.ToPtr(), raiseOnError, ffiOptions?.ToPtr() ?? IntPtr.Zero);
 
         // 4. Get a response and Handle it
         IntPtr response = await message;
@@ -137,16 +137,22 @@ internal virtual async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = n
     protected static readonly Version DefaultServerVersion = new(8, 0, 0);
     #endregion protected fields
 
+    #region internal fields
+    /// Raw pointer to the underlying native client.
+    internal IntPtr ClientPointer;
+    internal readonly MessageContainer MessageContainer;
+    #endregion internal fields
+
     #region private methods
     private void SuccessCallback(ulong index, IntPtr ptr) =>
         // Work needs to be offloaded from the calling thread, because otherwise we might starve the client's thread pool.
-        Task.Run(() => _messageContainer.GetMessage((int)index).SetResult(ptr));
+        Task.Run(() => MessageContainer.GetMessage((int)index).SetResult(ptr));
 
     private void FailureCallback(ulong index, IntPtr strPtr, RequestErrorType errType)
     {
         string str = Marshal.PtrToStringAnsi(strPtr)!;
         // Work needs to be offloaded from the calling thread, because otherwise we might starve the client's thread pool.
-        _ = Task.Run(() => _messageContainer.GetMessage((int)index).SetException(Create(errType, str)));
+        _ = Task.Run(() => MessageContainer.GetMessage((int)index).SetException(Create(errType, str)));
     }
 
     ~BaseClient() => Dispose();
@@ -155,83 +161,6 @@ private void FailureCallback(ulong index, IntPtr strPtr, RequestErrorType errTyp
 
     protected abstract Task<Version> GetServerVersionAsync();
 
-    /// <summary>
-    /// Executes a cluster scan command with the given cursor and arguments.
-    /// </summary>
-    /// <param name="cursor">The cursor for the scan iteration.</param>
-    /// <param name="args">Additional arguments for the scan command.</param>
-    /// <returns>A tuple containing the next cursor and the keys found in this iteration.</returns>
-    protected async Task<(string cursor, ValkeyKey[] keys)> ClusterScanCommand(string cursor, string[] args)
-    {
-        var message = _messageContainer.GetMessageForCall();
-        IntPtr cursorPtr = Marshal.StringToHGlobalAnsi(cursor);
-
-        IntPtr[]? argPtrs = null;
-        IntPtr argsPtr = IntPtr.Zero;
-        IntPtr argLengthsPtr = IntPtr.Zero;
-
-        try
-        {
-            if (args.Length > 0)
-            {
-                // 1. Get a pointer to the array of argument string pointers.
-                // Example: if args = ["MATCH", "key*"], then argPtrs[0] points
-                // to "MATCH", argPtrs[1] points to "key*", and argsPtr points
-                // to the argsPtrs array.
-                argPtrs = [.. args.Select(Marshal.StringToHGlobalAnsi)];
-                argsPtr = Marshal.AllocHGlobal(IntPtr.Size * args.Length);
-                Marshal.Copy(argPtrs, 0, argsPtr, args.Length);
-
-                // 2. Get a pointer to an array of argument string lengths.
-                // Example: if args = ["MATCH", "key*"], then argLengths[0] = 5
-                // (length of "MATCH"), argLengths[1] = 4 (length of "key*"),
-                // and argLengthsPtr points to the argLengths array.
-                var argLengths = args.Select(arg => (ulong)arg.Length).ToArray();
-                argLengthsPtr = Marshal.AllocHGlobal(sizeof(ulong) * args.Length);
-                Marshal.Copy(argLengths.Select(l => (long)l).ToArray(), 0, argLengthsPtr, args.Length);
-            }
-
-            // Submit request to Rust and wait for response.
-            RequestClusterScanFfi(_clientPointer, (ulong)message.Index, cursorPtr, (ulong)args.Length, argsPtr, argLengthsPtr);
-            IntPtr response = await message;
-
-            try
-            {
-                var result = HandleResponse(response);
-                var array = (object[])result!;
-                var nextCursor = array[0]!.ToString()!;
-                var keys = ((object[])array[1]!).Select(k => new ValkeyKey(k!.ToString())).ToArray();
-                return (nextCursor, keys);
-            }
-            finally
-            {
-                FreeResponse(response);
-            }
-        }
-        finally
-        {
-            // Clean up args memory
-            if (argLengthsPtr != IntPtr.Zero)
-            {
-                Marshal.FreeHGlobal(argLengthsPtr);
-            }
-
-            if (argsPtr != IntPtr.Zero)
-            {
-                Marshal.FreeHGlobal(argsPtr);
-            }
-
-            if (argPtrs != null)
-            {
-                Array.ForEach(argPtrs, Marshal.FreeHGlobal);
-            }
-
-            // Clean up cursor in Rust
-            RemoveClusterScanCursorFfi(cursorPtr);
-            Marshal.FreeHGlobal(cursorPtr);
-        }
-    }
-
     [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
     private delegate void SuccessAction(ulong index, IntPtr ptr);
     [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
@@ -248,9 +177,6 @@ private void FailureCallback(ulong index, IntPtr strPtr, RequestErrorType errTyp
     /// and held in order to prevent the cost of marshalling on each function call.
     private readonly SuccessAction _successCallbackDelegate;
 
-    /// Raw pointer to the underlying native client.
-    private IntPtr _clientPointer;
-    private readonly MessageContainer _messageContainer;
     private readonly object _lock = new();
     private string _clientInfo = ""; // used to distinguish and identify clients during tests
 

sources/Valkey.Glide/GlideClusterClient.cs

@@ -1,12 +1,16 @@
 // Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
 
+using System.Runtime.InteropServices;
+
 using Valkey.Glide.Commands;
 using Valkey.Glide.Commands.Options;
 using Valkey.Glide.Internals;
 using Valkey.Glide.Pipeline;
 
 using static Valkey.Glide.ConnectionConfiguration;
 using static Valkey.Glide.Errors;
+using static Valkey.Glide.Internals.FFI;
+using static Valkey.Glide.Internals.ResponseHandler;
 using static Valkey.Glide.Pipeline.Options;
 using static Valkey.Glide.Route;
 
@@ -319,6 +323,83 @@ protected override async Task<Version> GetServerVersionAsync()
         return _serverVersion;
     }
 
+    /// <summary>
+    /// Executes a cluster scan command with the given cursor and arguments.
+    /// </summary>
+    /// <param name="cursor">The cursor for the scan iteration.</param>
+    /// <param name="args">Additional arguments for the scan command.</param>
+    /// <returns>A tuple containing the next cursor and the keys found in this iteration.</returns>
+    private async Task<(string cursor, ValkeyKey[] keys)> ClusterScanCommand(string cursor, string[] args)
+    {
+        var message = MessageContainer.GetMessageForCall();
+        IntPtr cursorPtr = Marshal.StringToHGlobalAnsi(cursor);
+
+        IntPtr[]? argPtrs = null;
+        IntPtr argsPtr = IntPtr.Zero;
+        IntPtr argLengthsPtr = IntPtr.Zero;
+
+        try
+        {
+            if (args.Length > 0)
+            {
+                // 1. Get a pointer to the array of argument string pointers.
+                // Example: if args = ["MATCH", "key*"], then argPtrs[0] points
+                // to "MATCH", argPtrs[1] points to "key*", and argsPtr points
+                // to the argsPtrs array.
+                argPtrs = [.. args.Select(Marshal.StringToHGlobalAnsi)];
+                argsPtr = Marshal.AllocHGlobal(IntPtr.Size * args.Length);
+                Marshal.Copy(argPtrs, 0, argsPtr, args.Length);
+
+                // 2. Get a pointer to an array of argument string lengths.
+                // Example: if args = ["MATCH", "key*"], then argLengths[0] = 5
+                // (length of "MATCH"), argLengths[1] = 4 (length of "key*"),
+                // and argLengthsPtr points to the argLengths array.
+                var argLengths = args.Select(arg => (ulong)arg.Length).ToArray();
+                argLengthsPtr = Marshal.AllocHGlobal(sizeof(ulong) * args.Length);
+                Marshal.Copy(argLengths.Select(l => (long)l).ToArray(), 0, argLengthsPtr, args.Length);
+            }
+
+            // Submit request to Rust and wait for response.
+            RequestClusterScanFfi(ClientPointer, (ulong)message.Index, cursorPtr, (ulong)args.Length, argsPtr, argLengthsPtr);
+            IntPtr response = await message;
+
+            try
+            {
+                var result = HandleResponse(response);
+                var array = (object[])result!;
+                var nextCursor = array[0]!.ToString()!;
+                var keys = ((object[])array[1]!).Select(k => new ValkeyKey(k!.ToString())).ToArray();
+                return (nextCursor, keys);
+            }
+            finally
+            {
+                FreeResponse(response);
+            }
+        }
+        finally
+        {
+            // Clean up args memory
+            if (argLengthsPtr != IntPtr.Zero)
+            {
+                Marshal.FreeHGlobal(argLengthsPtr);
+            }
+
+            if (argsPtr != IntPtr.Zero)
+            {
+                Marshal.FreeHGlobal(argsPtr);
+            }
+
+            if (argPtrs != null)
+            {
+                Array.ForEach(argPtrs, Marshal.FreeHGlobal);
+            }
+
+            // Clean up cursor in Rust
+            RemoveClusterScanCursorFfi(cursorPtr);
+            Marshal.FreeHGlobal(cursorPtr);
+        }
+    }
+
     /// <summary>
     /// Iterates incrementally over keys in the cluster.
     /// </summary>