C#: batch compatibility (valkey-io#4371)

Signed-off-by: Yury-Fridlyand <yury.fridlyand@improving.com>
Co-authored-by: Edward Liang <76571219+edlng@users.noreply.github.com>

Author: Yury-Fridlyand

csharp/sources/Valkey.Glide/Abstract/ConnectionMultiplexer.cs

@@ -2,7 +2,6 @@
 
 using System.Net;
 
-using Valkey.Glide.Commands;
 using Valkey.Glide.Internals;
 
 using static Valkey.Glide.Commands.Options.InfoOptions;

csharp/sources/Valkey.Glide/Abstract/Database.cs

@@ -1,6 +1,5 @@
 // Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
 
-using Valkey.Glide.Commands;
 using Valkey.Glide.Commands.Options;
 using Valkey.Glide.Internals;
 
@@ -17,9 +16,21 @@ internal class DatabaseImpl : GlideClient, IDatabase
             ? await Command(Request.Info(sections), Route.Random)
             : await base.Info(sections);
 
+    public IBatch CreateBatch(object? asyncState = null)
+    {
+        Utils.Requires<ArgumentException>(asyncState is null, "Async state is not supported by GLIDE");
+        return new ValkeyBatch(this);
+    }
+
+    public ITransaction CreateTransaction(object? asyncState = null)
+    {
+        Utils.Requires<ArgumentException>(asyncState is null, "Async state is not supported by GLIDE");
+        return new ValkeyTransaction(this);
+    }
+
     internal readonly bool IsCluster;
 
-    private DatabaseImpl(bool isCluster) { IsCluster = isCluster; }
+    protected DatabaseImpl(bool isCluster) { IsCluster = isCluster; }
 
     public static async Task<DatabaseImpl> Create(BaseClientConfiguration config)
         => await CreateClient(config, () => new DatabaseImpl(config is ClusterClientConfiguration));

csharp/sources/Valkey.Glide/Abstract/IBatch.cs

@@ -0,0 +1,22 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+namespace Valkey.Glide;
+
+/// <summary>
+/// Represents a block of operations that will be sent to the server together.
+/// This can be useful to reduce packet fragmentation on slow connections - it
+/// can improve the time to get *all* the operations processed, with the trade-off
+/// of a slower time to get the *first* operation processed; this is usually
+/// a good thing. Unless this batch is a <b>transaction</b>, there is no guarantee
+/// that these operations will be processed either contiguously or atomically by the server.<br />
+/// See also <see cref="Pipeline.Batch" /> and  <see cref="Pipeline.ClusterBatch" />.
+/// </summary>
+public interface IBatch : IDatabaseAsync
+{
+    /// <summary>
+    /// Execute the batch operation, sending all queued commands to the server.
+    /// Note that this operation is neither synchronous nor truly asynchronous - it simply enqueues the buffered messages.
+    /// To check on completion, you should check the individual responses.
+    /// </summary>
+    void Execute();
+}

csharp/sources/Valkey.Glide/Abstract/IConnectionMultiplexer.cs

@@ -2,8 +2,6 @@
 
 using System.Net;
 
-using Valkey.Glide.Commands;
-
 namespace Valkey.Glide;
 
 /// <summary>

csharp/sources/Valkey.Glide/Abstract/IDatabase.cs

@@ -0,0 +1,26 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+namespace Valkey.Glide;
+
+/// <summary>
+/// Describes functionality that is common to both standalone and cluster servers.<br />
+/// See also <see cref="GlideClient" /> and <see cref="GlideClusterClient" />.
+/// </summary>
+public interface IDatabase : IDatabaseAsync
+{
+    /// <summary>
+    /// Allows creation of a group of operations that will be sent to the server as a single unit,
+    /// but which may or may not be processed on the server contiguously.
+    /// </summary>
+    /// <param name="asyncState">The async state is not supported by GLIDE.</param>
+    /// <returns>The created batch.</returns>
+    IBatch CreateBatch(object? asyncState = null);
+
+    /// <summary>
+    /// Allows creation of a group of operations that will be sent to the server as a single unit,
+    /// and processed on the server as a single unit.
+    /// </summary>
+    /// <param name="asyncState">The async state is not supported by GLIDE.</param>
+    /// <returns>The created transaction.</returns>
+    ITransaction CreateTransaction(object? asyncState = null);
+}

csharp/sources/Valkey.Glide/Abstract/IDatabaseAsync.cs

@@ -0,0 +1,8 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+using Valkey.Glide.Commands;
+
+namespace Valkey.Glide;
+
+public interface IDatabaseAsync : IStringBaseCommands, IGenericCommands, IServerManagementCommands
+{ }

csharp/sources/Valkey.Glide/Abstract/ITransaction.cs

@@ -0,0 +1,45 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+namespace Valkey.Glide;
+
+/// <summary>
+/// Represents a group of operations that will be sent to the server as a single unit,
+/// and processed on the server as a single unit. Transactions can also include constraints
+/// (implemented via <c>WATCH</c>), but note that constraint checking involves will (very briefly)
+/// block the connection, since the transaction cannot be correctly committed (<c>EXEC</c>),
+/// aborted (<c>DISCARD</c>) or not applied in the first place (<c>UNWATCH</c>) until the responses from
+/// the constraint checks have arrived.
+/// See also <see cref="Pipeline.Batch" /> and  <see cref="Pipeline.ClusterBatch" />.
+/// </summary>
+/// <remarks>
+/// <para>Note that on a cluster, it may be required that all keys involved in the transaction (including constraints) are in the same hash-slot.</para>
+/// <para><seealso href="https://valkey.io/topics/transactions/"/></para>
+/// </remarks>
+public interface ITransaction : IBatch
+{
+    /// <summary>
+    /// Adds a precondition for this transaction.
+    /// </summary>
+    /// <param name="condition">The condition to add to the transaction.</param>
+    //ConditionResult AddCondition(Condition condition);
+
+    /// <summary>
+    /// Execute the batch operation, sending all queued commands to the server.
+    /// </summary>
+    /// <param name="flags">Command flags are not supported by GLIDE.</param>
+    /// <returns>
+    /// <see langword="true" /> if a transaction was applied or
+    /// <see langword="false" /> if a transaction failed due to a <c>WATCH</c> command.
+    /// </returns>
+    bool Execute(CommandFlags flags = CommandFlags.None);
+
+    /// <summary>
+    /// Execute the batch operation, sending all queued commands to the server.
+    /// </summary>
+    /// <param name="flags">Command flags are not supported by GLIDE.</param>
+    /// <returns>
+    /// <see langword="true" /> if a transaction was applied or
+    /// <see langword="false" /> if a transaction failed due to a <c>WATCH</c> command.
+    /// </returns>
+    Task<bool> ExecuteAsync(CommandFlags flags = CommandFlags.None);
+}

csharp/sources/Valkey.Glide/Abstract/ValkeyBatch.cs

@@ -0,0 +1,34 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+using Valkey.Glide.Internals;
+using Valkey.Glide.Pipeline;
+
+namespace Valkey.Glide;
+
+internal class ValkeyBatch(BaseClient client) : DatabaseImpl(false), IBatch
+{
+    private readonly List<ICmd> _commands = [];
+    protected TaskCompletionSource<object?[]?> _tcs = new();
+    private readonly BaseClient _client = client;
+    protected bool _isAtomic = false;
+
+    internal override async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = null)
+    {
+        int idx = _commands.Count;
+        _commands.Add(command);
+        // wait for the batch to be executed and then pick command's result
+        return await _tcs.Task.ContinueWith(task => (T)task.Result![idx]!);
+        // TODO what if transactoin failed?
+    }
+
+    protected async Task ExecuteImpl()
+    {
+        Batch b = new(_isAtomic);
+        b.Commands.AddRange(_commands);
+
+        object?[]? res = await _client.Batch(b, true);
+        _tcs.SetResult(res);
+    }
+
+    public void Execute() => ExecuteImpl().GetAwaiter().GetResult();
+}

csharp/sources/Valkey.Glide/Abstract/ValkeyTransaction.cs

@@ -0,0 +1,21 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+namespace Valkey.Glide;
+
+internal class ValkeyTransaction : ValkeyBatch, ITransaction
+{
+    public ValkeyTransaction(BaseClient client) : base(client)
+    {
+        _isAtomic = true;
+    }
+
+    public bool Execute(CommandFlags flags = CommandFlags.None)
+        => ExecuteAsync(flags).GetAwaiter().GetResult();
+
+    public async Task<bool> ExecuteAsync(CommandFlags flags = CommandFlags.None)
+    {
+        Utils.Requires<NotImplementedException>(flags == CommandFlags.None, "Command flags are not supported by GLIDE");
+        await ExecuteImpl();
+        return _tcs.Task.Result is not null;
+    }
+}

csharp/sources/Valkey.Glide/BaseClient.cs

@@ -66,14 +66,11 @@ protected BaseClient()
 
     protected internal delegate T ResponseHandler<T>(IntPtr response);
 
-    /// <summary>
-    /// </summary>
     /// <typeparam name="R">Type received from server.</typeparam>
     /// <typeparam name="T">Type we return to the user.</typeparam>
     /// <param name="command"></param>
     /// <param name="route"></param>
-    /// <returns></returns>
-    internal async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = null)
+    internal virtual async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = null)
     {
         // 1. Create Cmd which wraps CmdInfo and manages all memory allocations
         using Cmd cmd = command.ToFfi();
@@ -99,7 +96,7 @@ internal async Task<T> Command<R, T>(Cmd<R, T> command, Route? route = null)
         // All memory allocated is auto-freed by `using` operator
     }
 
-    protected async Task<object?[]?> Batch<T>(BaseBatch<T> batch, bool raiseOnError, BaseBatchOptions? options = null) where T : BaseBatch<T>
+    internal async Task<object?[]?> Batch<T>(BaseBatch<T> batch, bool raiseOnError, BaseBatchOptions? options = null) where T : BaseBatch<T>
     {
         // 1. Allocate memory for batch, which allocates all nested Cmds
         using FFI.Batch ffiBatch = batch.ToFFI();