Add existential execute to ValkeyClientProtocol (#235)

* Add existential execute to ValkeyClientProtocol

Signed-off-by: Adam Fowler <[email protected]>

* remove `@inlinable` from protocol

Signed-off-by: Adam Fowler <[email protected]>

---------

Signed-off-by: Adam Fowler <[email protected]>

Author: adam-fowler

Sources/Valkey/Cluster/ValkeyClusterClient.swift

@@ -207,7 +207,7 @@ public final class ValkeyClusterClient: Sendable {
     @inlinable
     public func execute<each Command: ValkeyCommand>(
         _ commands: repeat each Command
-    ) async throws -> sending (repeat Result<(each Command).Response, Error>) {
+    ) async -> sending (repeat Result<(each Command).Response, Error>) {
         func convert<Response: RESPTokenDecodable>(_ result: Result<RESPToken, Error>, to: Response.Type) -> Result<Response, Error> {
             result.flatMap {
                 do {
@@ -217,7 +217,7 @@ public final class ValkeyClusterClient: Sendable {
                 }
             }
         }
-        let results = try await self.execute([any ValkeyCommand](commands: repeat each commands))
+        let results = await self.execute([any ValkeyCommand](commands: repeat each commands))
         var index = AutoIncrementingInteger()
         return (repeat convert(results[index.next()], to: (each Command).Response.self))
     }
@@ -259,40 +259,44 @@ public final class ValkeyClusterClient: Sendable {
     @inlinable
     public func execute(
         _ commands: [any ValkeyCommand]
-    ) async throws -> sending [Result<RESPToken, Error>] {
+    ) async -> sending [Result<RESPToken, Error>] {
         guard commands.count > 0 else { return [] }
         // get a list of nodes and the commands that should be run on them
-        let nodes = try await self.splitCommandsAcrossNodes(commands: commands)
-        // if this list has one element, then just run the pipeline on that single node
-        if nodes.count == 1 {
-            do {
-                return try await self.execute(node: nodes[nodes.startIndex].node, commands: commands)
-            } catch {
-                return .init(repeating: .failure(error), count: commands.count)
+        do {
+            let nodes = try await self.splitCommandsAcrossNodes(commands: commands)
+            // if this list has one element, then just run the pipeline on that single node
+            if nodes.count == 1 {
+                do {
+                    return try await self.execute(node: nodes[nodes.startIndex].node, commands: commands)
+                } catch {
+                    return .init(repeating: .failure(error), count: commands.count)
+                }
             }
-        }
-        return await withTaskGroup(of: NodePipelineResult.self) { group in
-            // run generated pipelines concurrently
-            for node in nodes {
-                let indices = node.commandIndices
-                group.addTask {
-                    do {
-                        let results = try await self.execute(node: node.node, commands: IndexedSubCollection(commands, indices: indices))
-                        return .init(indices: indices, results: results)
-                    } catch {
-                        return NodePipelineResult(indices: indices, results: .init(repeating: .failure(error), count: indices.count))
+            return await withTaskGroup(of: NodePipelineResult.self) { group in
+                // run generated pipelines concurrently
+                for node in nodes {
+                    let indices = node.commandIndices
+                    group.addTask {
+                        do {
+                            let results = try await self.execute(node: node.node, commands: IndexedSubCollection(commands, indices: indices))
+                            return .init(indices: indices, results: results)
+                        } catch {
+                            return NodePipelineResult(indices: indices, results: .init(repeating: .failure(error), count: indices.count))
+                        }
                     }
                 }
-            }
-            var results = [Result<RESPToken, Error>](repeating: .failure(ValkeyClusterError.pipelinedResultNotReturned), count: commands.count)
-            // get results for each node
-            while let taskResult = await group.next() {
-                precondition(taskResult.indices.count == taskResult.results.count)
-                for index in 0..<taskResult.indices.count {
-                    results[taskResult.indices[index]] = taskResult.results[index]
+                var results = [Result<RESPToken, Error>](repeating: .failure(ValkeyClusterError.pipelinedResultNotReturned), count: commands.count)
+                // get results for each node
+                while let taskResult = await group.next() {
+                    precondition(taskResult.indices.count == taskResult.results.count)
+                    for index in 0..<taskResult.indices.count {
+                        results[taskResult.indices[index]] = taskResult.results[index]
+                    }
                 }
+                return results
             }
-            return results
+        } catch {
+            return .init(repeating: .failure(error), count: commands.count)
         }
     }
 

Sources/Valkey/ValkeyClientProtocol.swift

@@ -7,9 +7,23 @@
 //
 /// A type that provides the ability to send a Valkey command and get a response.
 @available(valkeySwift 1.0, *)
-public protocol ValkeyClientProtocol {
+public protocol ValkeyClientProtocol: Sendable {
     /// Send RESP command to Valkey connection
     /// - Parameter command: ValkeyCommand structure
     /// - Returns: The command response as defined in the ValkeyCommand
     func execute<Command: ValkeyCommand>(_ command: Command) async throws -> Command.Response
+
+    /// Pipeline a series of commands to Valkey connection
+    ///
+    /// Once all the responses for the commands have been received the function returns
+    /// an array of RESPToken Results, one for each command.
+    ///
+    /// This is an alternative version of the pipelining function ``ValkeyClient/execute(_:)->(_,_)``
+    /// that allows for a collection of ValkeyCommands. It provides more flexibility but
+    /// is more expensive to run and the command responses are returned as ``RESPToken``
+    /// instead of the response type for the command.
+    ///
+    /// - Parameter commands: Collection of ValkeyCommands
+    /// - Returns: Array holding the RESPToken responses of all the commands
+    func execute(_ commands: [any ValkeyCommand]) async -> sending [Result<RESPToken, Error>]
 }

Tests/ClusterIntegrationTests/ClusterIntegrationTests.swift

@@ -615,7 +615,7 @@ struct ClusterIntegrationTests {
                     var commands: [any ValkeyCommand] = .init()
                     commands.append(SET(key, value: "cluster pipeline test"))
                     commands.append(GET(key))
-                    let results = try await client.execute(commands)
+                    let results = await client.execute(commands)
                     let response = try results[1].get().decode(as: String.self)
                     #expect(response == "cluster pipeline test")
                 }
@@ -638,7 +638,7 @@ struct ClusterIntegrationTests {
                     commands.append(GET(key))
                     commands.append(DEL(keys: [key]))
                 }
-                let results = try await client.execute(commands)
+                let results = await client.execute(commands)
                 let response = try results[1].get().decode(as: String.self)
                 #expect(response == "0")
                 let response2 = try results[7].get().decode(as: String.self)
@@ -655,7 +655,7 @@ struct ClusterIntegrationTests {
             let firstNodePort = clusterFirstNodePort ?? 6379
             try await ClusterIntegrationTests.withValkeyCluster([(host: firstNodeHostname, port: firstNodePort)], logger: logger) {
                 client in
-                let results = try await client.execute(
+                let results = await client.execute(
                     SET("test1", value: "1"),
                     GET("test1"),
                     DEL(keys: ["test1"]),

Tests/IntegrationTests/ClientIntegrationTests.swift

@@ -203,6 +203,25 @@ struct ClientIntegratedTests {
         }
     }
 
+    @Test
+    @available(valkeySwift 1.0, *)
+    func testPipelinedProtocolSetGet() async throws {
+        @Sendable func setGet(_ client: some ValkeyClientProtocol, key: ValkeyKey) async throws {
+            var commands: [any ValkeyCommand] = []
+            commands.append(SET(key, value: "Pipelined Hello"))
+            commands.append(GET(key))
+            let responses = await client.execute(commands)
+            try #expect(responses[1].get().decode(as: String.self) == "Pipelined Hello")
+        }
+        var logger = Logger(label: "Valkey")
+        logger.logLevel = .debug
+        try await withValkeyConnection(.hostname(valkeyHostname, port: 6379), logger: logger) { connection in
+            try await withKey(connection: connection) { key in
+                try await setGet(connection, key: key)
+            }
+        }
+    }
+
     @Test
     @available(valkeySwift 1.0, *)
     func testPipelinedSetGetClient() async throws {