Call CLIENT SETINFO on initialisation of connection (#170)

* Fire and forget CLIENT.SETINFO on connection initialization

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

* ignore -> forget

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

* Add version.swift with library version/name strings

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

* Add test to check library name and version are being set

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

* Add support for returning dummy value from CLIENT SETINFO command

Moved non-custom command handling to channel handler

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

* Add BenchmarkCommandHandler parameter to makeLocalServer

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

---------

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

Author: adam-fowler

Benchmarks/ValkeyBenchmarks/Shared.swift

@@ -33,29 +33,27 @@ let defaultMetrics: [BenchmarkMetric] =
         .throughput,
     ]
 
-func makeLocalServer() async throws -> Channel {
-    struct GetHandler: BenchmarkCommandHandler {
-        static let expectedCommand = RESPToken.Value.bulkString(ByteBuffer(string: "GET"))
-        static let response = ByteBuffer(string: "$3\r\nBar\r\n")
-        func handle(command: RESPToken.Value, parameters: RESPToken.Array.Iterator, write: (ByteBuffer) -> Void) {
-            switch command {
-            case Self.expectedCommand:
-                write(Self.response)
-            case .bulkString(ByteBuffer(string: "PING")):
-                write(ByteBuffer(string: "$4\r\nPONG\r\n"))
-            case .bulkString(let string):
-                fatalError("Unexpected command: \(String(buffer: string))")
-            default:
-                fatalError("Unexpected value: \(command)")
-            }
+struct BenchmarkGetHandler: BenchmarkCommandHandler {
+    static let expectedCommand = RESPToken.Value.bulkString(ByteBuffer(string: "GET"))
+    static let response = ByteBuffer(string: "$3\r\nBar\r\n")
+    func handle(command: RESPToken.Value, parameters: RESPToken.Array.Iterator, write: (ByteBuffer) -> Void) {
+        switch command {
+        case Self.expectedCommand:
+            write(Self.response)
+        case .bulkString(let string):
+            fatalError("Unexpected command: \(String(buffer: string))")
+        default:
+            fatalError("Unexpected value: \(command)")
         }
     }
-    return try await ServerBootstrap(group: NIOSingletons.posixEventLoopGroup)
+}
+func makeLocalServer(commandHandler: some BenchmarkCommandHandler = BenchmarkGetHandler()) async throws -> Channel {
+    try await ServerBootstrap(group: NIOSingletons.posixEventLoopGroup)
         .serverChannelOption(.socketOption(.so_reuseaddr), value: 1)
         .childChannelInitializer { channel in
             do {
                 try channel.pipeline.syncOperations.addHandler(
-                    ValkeyServerChannelHandler(commandHandler: GetHandler())
+                    ValkeyServerChannelHandler(commandHandler: commandHandler)
                 )
                 return channel.eventLoop.makeSucceededVoidFuture()
             } catch {
@@ -66,7 +64,7 @@ func makeLocalServer() async throws -> Channel {
         .get()
 }
 
-protocol BenchmarkCommandHandler {
+protocol BenchmarkCommandHandler: Sendable {
     func handle(command: RESPToken.Value, parameters: RESPToken.Array.Iterator, write: (ByteBuffer) -> Void)
 }
 
@@ -78,6 +76,11 @@ final class ValkeyServerChannelHandler<Handler: BenchmarkCommandHandler>: Channe
     private var decoder = NIOSingleStepByteToMessageProcessor(RESPTokenDecoder())
     private let helloCommand = RESPToken.Value.bulkString(ByteBuffer(string: "HELLO"))
     private let helloResponse = ByteBuffer(string: "%1\r\n+server\r\n+fake\r\n")
+    private let pingCommand = RESPToken.Value.bulkString(ByteBuffer(string: "PING"))
+    private let pongResponse = ByteBuffer(string: "$4\r\nPONG\r\n")
+    private let clientCommand = RESPToken.Value.bulkString(ByteBuffer(string: "CLIENT"))
+    private let setInfoSubCommand = RESPToken.Value.bulkString(ByteBuffer(string: "SETINFO"))
+    private let okResponse = ByteBuffer(string: "+2OK\r\n")
     private let commandHandler: Handler
 
     init(commandHandler: Handler) {
@@ -102,6 +105,20 @@ final class ValkeyServerChannelHandler<Handler: BenchmarkCommandHandler>: Channe
         case helloCommand:
             context.writeAndFlush(self.wrapOutboundOut(helloResponse), promise: nil)
 
+        case pingCommand:
+            context.writeAndFlush(self.wrapOutboundOut(pongResponse), promise: nil)
+
+        case clientCommand:
+            var subCommandIterator = iterator
+            switch subCommandIterator.next()?.value {
+            case setInfoSubCommand:
+                context.writeAndFlush(self.wrapOutboundOut(okResponse), promise: nil)
+            default:
+                commandHandler.handle(command: command, parameters: iterator) {
+                    context.writeAndFlush(self.wrapOutboundOut($0), promise: nil)
+                }
+            }
+
         default:
             commandHandler.handle(command: command, parameters: iterator) {
                 context.writeAndFlush(self.wrapOutboundOut($0), promise: nil)

Sources/Valkey/Connection/ValkeyChannelHandler+stateMachine.swift

@@ -225,7 +225,7 @@ extension ValkeyChannelHandler {
                 case .nio(let promise):
                     self = .connected(state)
                     return .waitForPromise(promise)
-                case .swift:
+                case .swift, .forget:
                     preconditionFailure("Connected state cannot be setup with a Swift continuation")
                 }
             case .active(let state):

Sources/Valkey/Connection/ValkeyChannelHandler.swift

@@ -20,13 +20,16 @@ import NIOCore
 enum ValkeyPromise<T: Sendable>: Sendable {
     case nio(EventLoopPromise<T>)
     case swift(CheckedContinuation<T, any Error>)
+    case forget
 
     func succeed(_ t: T) {
         switch self {
         case .nio(let eventLoopPromise):
             eventLoopPromise.succeed(t)
         case .swift(let checkedContinuation):
             checkedContinuation.resume(returning: t)
+        case .forget:
+            break
         }
     }
 
@@ -36,6 +39,8 @@ enum ValkeyPromise<T: Sendable>: Sendable {
             eventLoopPromise.fail(e)
         case .swift(let checkedContinuation):
             checkedContinuation.resume(throwing: e)
+        case .forget:
+            break
         }
     }
 }
@@ -154,6 +159,33 @@ final class ValkeyChannelHandler: ChannelInboundHandler {
         }
     }
 
+    /// Write valkey command/commands to channel
+    /// - Parameters:
+    ///   - request: Valkey command request
+    ///   - promise: Promise to fulfill when command is complete
+    @inlinable
+    func writeAndForget<Command: ValkeyCommand>(command: Command, requestID: Int) {
+        self.eventLoop.assertInEventLoop()
+        let pendingCommand = PendingCommand(
+            promise: .forget,
+            requestID: requestID,
+            deadline: .now() + self.configuration.commandTimeout
+        )
+        switch self.stateMachine.sendCommand(pendingCommand) {
+        case .sendCommand(let context):
+            self.encoder.reset()
+            command.encode(into: &self.encoder)
+            let buffer = self.encoder.buffer
+            context.writeAndFlush(self.wrapOutboundOut(buffer), promise: nil)
+            if self.deadlineCallback == nil {
+                self.scheduleDeadlineCallback(deadline: .now() + self.configuration.commandTimeout)
+            }
+
+        case .throwError:
+            break
+        }
+    }
+
     @usableFromInline
     func write(request: ValkeyRequest) {
         self.eventLoop.assertInEventLoop()

Sources/Valkey/Connection/ValkeyConnection.swift

@@ -132,7 +132,7 @@ public final actor ValkeyConnection: ValkeyClientProtocol, Sendable {
                 }
             }
         let connection = try await future.get()
-        try await connection.waitOnActive()
+        try await connection.initialHandshake()
         return connection
     }
 
@@ -144,8 +144,10 @@ public final actor ValkeyConnection: ValkeyClientProtocol, Sendable {
         self.channel.close(mode: .all, promise: nil)
     }
 
-    func waitOnActive() async throws {
+    func initialHandshake() async throws {
         try await self.channelHandler.waitOnActive().get()
+        self.executeAndForget(command: CLIENT.SETINFO(attr: .libname(valkeySwiftLibraryName)))
+        self.executeAndForget(command: CLIENT.SETINFO(attr: .libver(valkeySwiftLibraryVersion)))
     }
 
     /// Send RESP command to Valkey connection
@@ -172,6 +174,10 @@ public final actor ValkeyConnection: ValkeyClientProtocol, Sendable {
         }
     }
 
+    func executeAndForget<Command: ValkeyCommand>(command: Command) {
+        self.channelHandler.writeAndForget(command: command, requestID: Self.requestIDGenerator.next())
+    }
+
     /// Pipeline a series of commands to Valkey connection
     ///
     /// Once all the responses for the commands have been received the function returns

Sources/Valkey/ValkeyConnectionFactory.swift

@@ -94,7 +94,7 @@ package final class ValkeyConnectionFactory: Sendable {
                     logger: logger
                 )
             }.get()
-            try await connection.waitOnActive()
+            try await connection.initialHandshake()
             return connection
         }
     }

Sources/Valkey/Version.swift

@@ -0,0 +1,16 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the valkey-swift open source project
+//
+// Copyright (c) 2025 the valkey-swift project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of valkey-swift project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+package let valkeySwiftLibraryName = "valkey-swift"
+package let valkeySwiftLibraryVersion = "0.1.0"

Tests/IntegrationTests/ValkeyTests.swift

@@ -859,4 +859,16 @@ struct GeneratedCommands {
             }
         }
     }
+
+    @available(valkeySwift 1.0, *)
+    @Test
+    func testClientInfo() async throws {
+        var logger = Logger(label: "Valkey")
+        logger.logLevel = .trace
+        try await withValkeyClient(.hostname(valkeyHostname, port: 6379), logger: logger) { client in
+            let clients = try await String(buffer: client.clientList())
+            #expect(clients.firstRange(of: "lib-name=\(valkeySwiftLibraryName)") != nil)
+            #expect(clients.firstRange(of: "lib-ver=\(valkeySwiftLibraryVersion)") != nil)
+        }
+    }
 }