ValkeyClient.pipeline (#104)

Author: adam-fowler

Sources/Valkey/ValkeyClient.swift

@@ -169,6 +169,26 @@ extension ValkeyClient: ValkeyConnectionProtocol {
     }
 }
 
+extension ValkeyClient {
+    /// Pipeline a series of commands to Valkey connection
+    ///
+    /// This function will only return once it has the results of all the commands sent
+    /// - Parameter commands: Parameter pack of ValkeyCommands
+    /// - Returns: Parameter pack holding the results of all the commands
+    @inlinable
+    public func pipeline<each Command: ValkeyCommand>(
+        _ commands: repeat each Command
+    ) async -> sending (repeat Result<(each Command).Response, Error>) {
+        do {
+            return try await self.withConnection { connection in
+                await connection.pipeline(repeat (each commands))
+            }
+        } catch {
+            return (repeat Result<(each Command).Response, Error>.failure(error))
+        }
+    }
+}
+
 #if ServiceLifecycleSupport
 extension ValkeyClient: Service {}
 #endif  // ServiceLifecycle

Tests/IntegrationTests/ValkeyTests.swift

@@ -165,6 +165,21 @@ struct GeneratedCommands {
         }
     }
 
+    @Test
+    func testPipelinedSetGetClient() async throws {
+        var logger = Logger(label: "Valkey")
+        logger.logLevel = .debug
+        try await withValkeyClient(.hostname(valkeyHostname, port: 6379), logger: logger) { client in
+            try await withKey(connection: client) { key in
+                let responses = await client.pipeline(
+                    SET(key: key, value: "Pipelined Hello"),
+                    GET(key: key)
+                )
+                try #expect(responses.1.get()?.decode(as: String.self) == "Pipelined Hello")
+            }
+        }
+    }
+
     @Test
     func testTransactionSetIncrGet() async throws {
         var logger = Logger(label: "Valkey")