Add ValkeyConnection.withConnection() (#155)

* Add ValkeyConnection.withConnection()

Replacing ValkeyConnection.connection()

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

* Remove client name parameter

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

---------

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

Author: adam-fowler

Benchmarks/ValkeyBenchmarks/ValkeyConnectionBenchmark.swift

@@ -32,21 +32,18 @@ func makeConnectionGETBenchmark() -> Benchmark? {
     return Benchmark("Connection: GET benchmark", configuration: .init(metrics: defaultMetrics, scalingFactor: .kilo)) { benchmark in
         let port = serverMutex.withLock { $0 }!.localAddress!.port!
         let logger = Logger(label: "test")
-        let connection = try await ValkeyConnection.connect(
+        try await ValkeyConnection.withConnection(
             address: .hostname("127.0.0.1", port: port),
-            connectionID: 1,
             configuration: .init(),
             logger: logger
-        )
-
-        benchmark.startMeasurement()
-        for _ in benchmark.scaledIterations {
-            let foo = try await connection.get("foo")
-            precondition(foo.map { String(buffer: $0) } == "Bar")
+        ) { connection in
+            benchmark.startMeasurement()
+            for _ in benchmark.scaledIterations {
+                let foo = try await connection.get("foo")
+                precondition(foo.map { String(buffer: $0) } == "Bar")
+            }
+            benchmark.stopMeasurement()
         }
-        benchmark.stopMeasurement()
-
-        connection.close()
     } setup: {
         let server = try await makeLocalServer()
         serverMutex.withLock { $0 = server }

Sources/Valkey/Connection/ValkeyConnection.swift

@@ -59,42 +59,74 @@ public final actor ValkeyConnection: ValkeyConnectionProtocol, Sendable {
         self.isClosed = .init(false)
     }
 
+    /// Connect to Valkey database and run operation using connection and then close
+    /// connection.
+    ///
+    /// To avoid the cost of acquiring the connection and then closing it, it is always
+    /// preferable to use ``ValkeyClient/withConnection(isolation:operation:)`` which
+    /// uses a persistent connection pool to provide connections to your Valkey database.
+    ///
+    /// - Parameters:
+    ///   - address: Internet address of database
+    ///   - configuration: Configuration of Valkey connection
+    ///   - eventLoop: EventLoop to run connection on
+    ///   - logger: Logger for connection
+    ///   - isolation: Actor isolation
+    ///   - operation: Closure handling Valkey connection
+    /// - Returns: Return value of operation closure
+    public static func withConnection<Value>(
+        address: ValkeyServerAddress,
+        configuration: ValkeyConnectionConfiguration = .init(),
+        eventLoop: EventLoop = MultiThreadedEventLoopGroup.singleton.any(),
+        logger: Logger,
+        isolation: isolated (any Actor)? = #isolation,
+        operation: (ValkeyConnection) async throws -> sending Value
+    ) async throws -> sending Value {
+        let connection = try await connect(
+            address: address,
+            connectionID: 0,
+            configuration: configuration,
+            eventLoop: eventLoop,
+            logger: logger
+        )
+        defer {
+            connection.close()
+        }
+        return try await operation(connection)
+    }
+
     /// Connect to Valkey database and return connection
     ///
     /// - Parameters:
     ///   - address: Internet address of database
     ///   - connectionID: Connection identifier
-    ///   - name: Name of connection. The connection name is used in many of the `CLIENT` commands.
     ///   - configuration: Configuration of Valkey connection
     ///   - eventLoop: EventLoop to run connection on
     ///   - logger: Logger for connection
     /// - Returns: ValkeyConnection
-    public static func connect(
+    static func connect(
         address: ValkeyServerAddress,
         connectionID: ID,
-        name: String? = nil,
         configuration: ValkeyConnectionConfiguration,
         eventLoop: EventLoop = MultiThreadedEventLoopGroup.singleton.any(),
         logger: Logger
     ) async throws -> ValkeyConnection {
         let future =
             if eventLoop.inEventLoop {
-                self._makeClient(
+                self._makeConnection(
                     address: address,
                     connectionID: connectionID,
                     eventLoop: eventLoop,
                     configuration: configuration,
-                    clientName: name,
                     logger: logger
                 )
             } else {
                 eventLoop.flatSubmit {
-                    self._makeClient(
+                    self._makeConnection(
                         address: address,
                         connectionID: connectionID,
                         eventLoop: eventLoop,
                         configuration: configuration,
-                        clientName: name,
                         logger: logger
                     )
                 }
@@ -195,12 +227,11 @@ public final actor ValkeyConnection: ValkeyConnectionProtocol, Sendable {
     }
 
     /// Create Valkey connection and return channel connection is running on and the Valkey channel handler
-    private static func _makeClient(
+    private static func _makeConnection(
         address: ValkeyServerAddress,
         connectionID: ID,
         eventLoop: EventLoop,
         configuration: ValkeyConnectionConfiguration,
-        clientName: String?,
         logger: Logger
     ) -> EventLoopFuture<ValkeyConnection> {
         eventLoop.assertInEventLoop()

Sources/Valkey/Documentation.docc/Pipelining.md

@@ -2,8 +2,6 @@
 
 Sending multiple commands at once without waiting for the response of each command
 
-## Overview
-
 Valkey pipelining is a technique for improving performance by issuing multiple commands at once without waiting for the response to each individual command. Pipelining not only reduces the latency cost of waiting for the result of each command it also reduces the cost to the server as it reduces I/O costs. Multiple commands can be read with a single syscall, and multiple results are delivered with a single syscall. 
 
 ## Implementation

Tests/IntegrationTests/ValkeyTests.swift

@@ -68,6 +68,7 @@ struct GeneratedCommands {
             }
         }
     }
+
     @Test
     @available(valkeySwift 1.0, *)
     func testValkeyCommand() async throws {
@@ -95,6 +96,22 @@ struct GeneratedCommands {
         }
     }
 
+    @Test("Test ValkeyConnection.withConnection()")
+    @available(valkeySwift 1.0, *)
+    func testWithConnectionSetGet() async throws {
+        var logger = Logger(label: "Valkey")
+        logger.logLevel = .debug
+        try await ValkeyConnection.withConnection(address: .hostname(valkeyHostname, port: 6379), logger: logger) { connection in
+            try await withKey(connection: connection) { key in
+                try await connection.set(key, value: "Hello")
+                let response = try await connection.get(key).map { String(buffer: $0) }
+                #expect(response == "Hello")
+                let response2 = try await connection.get("sdf65fsdf").map { String(buffer: $0) }
+                #expect(response2 == nil)
+            }
+        }
+    }
+
     @Test
     @available(valkeySwift 1.0, *)
     func testSetGet() async throws {