Add fake server channel handler to test server close (#192)

Signed-off-by: Adam Fowler <adamfowler71@gmail.com>

Author: adam-fowler

Benchmarks/ValkeyBenchmarks/Shared.swift

@@ -51,13 +51,10 @@ func makeLocalServer(commandHandler: some BenchmarkCommandHandler = BenchmarkGet
     try await ServerBootstrap(group: NIOSingletons.posixEventLoopGroup)
         .serverChannelOption(.socketOption(.so_reuseaddr), value: 1)
         .childChannelInitializer { channel in
-            do {
+            channel.eventLoop.makeCompletedFuture {
                 try channel.pipeline.syncOperations.addHandler(
                     ValkeyServerChannelHandler(commandHandler: commandHandler)
                 )
-                return channel.eventLoop.makeSucceededVoidFuture()
-            } catch {
-                return channel.eventLoop.makeFailedFuture(error)
             }
         }
         .bind(host: "127.0.0.1", port: 0)

Tests/ValkeyTests/Utils/TestValkeyServerChannelHandler.swift

@@ -0,0 +1,83 @@
+//===----------------------------------------------------------------------===//
+//
+// 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
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+import Valkey
+
+// Fake Valkey server channel handler
+final class TestValkeyServerChannelHandler: ChannelInboundHandler {
+    typealias InboundIn = ByteBuffer
+    typealias OutboundOut = ByteBuffer
+
+    private var decoder = NIOSingleStepByteToMessageProcessor(RESPTokenDecoder())
+    private let commandHandler: (String, ArraySlice<String>, (ByteBuffer) -> Void) -> Void
+
+    static private let helloCommand = RESPToken.Value.bulkString(ByteBuffer(string: "HELLO"))
+    static private let helloResponse = ByteBuffer(string: "%1\r\n+server\r\n+fake\r\n")
+    static private let pingCommand = RESPToken.Value.bulkString(ByteBuffer(string: "PING"))
+    static private let pongResponse = ByteBuffer(string: "$4\r\nPONG\r\n")
+    static private let clientCommand = RESPToken.Value.bulkString(ByteBuffer(string: "CLIENT"))
+    static private let setInfoSubCommand = RESPToken.Value.bulkString(ByteBuffer(string: "SETINFO"))
+    static private let okResponse = ByteBuffer(string: "+2OK\r\n")
+
+    static private let response = ByteBuffer(string: "$3\r\nBar\r\n")
+    static func defaultHandler(command: String, parameters: ArraySlice<String>, write: (ByteBuffer) -> Void) {
+        guard command == "GET" else {
+            fatalError("Unexpected command: \(command)")
+        }
+        write(Self.response)
+    }
+
+    init(commandHandler: @escaping (String, ArraySlice<String>, (ByteBuffer) -> Void) -> Void = defaultHandler) {
+        self.commandHandler = commandHandler
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        try! self.decoder.process(buffer: self.unwrapInboundIn(data)) { token in
+            self.handleToken(context: context, token: token)
+        }
+    }
+
+    func handleToken(context: ChannelHandlerContext, token: RESPToken) {
+        guard let fullCommand = try? token.decode(as: [String].self) else {
+            fatalError()
+        }
+        guard let command = fullCommand.first else {
+            fatalError()
+        }
+        let parameters = fullCommand.dropFirst()
+        switch command {
+        case "HELLO":
+            context.writeAndFlush(self.wrapOutboundOut(Self.helloResponse), promise: nil)
+
+        case "PING":
+            context.writeAndFlush(self.wrapOutboundOut(Self.pongResponse), promise: nil)
+
+        case "CLIENT":
+            switch parameters.first {
+            case "SETINFO":
+                context.writeAndFlush(self.wrapOutboundOut(Self.okResponse), promise: nil)
+            default:
+                commandHandler(command, parameters) {
+                    context.writeAndFlush(self.wrapOutboundOut($0), promise: nil)
+                }
+            }
+
+        default:
+            commandHandler(command, parameters) {
+                context.writeAndFlush(self.wrapOutboundOut($0), promise: nil)
+            }
+        }
+    }
+}

Tests/ValkeyTests/ValkeyConnectionTests.swift

@@ -15,6 +15,7 @@
 import Logging
 import NIOCore
 import NIOEmbedded
+import NIOPosix
 import Testing
 
 @testable import Valkey
@@ -449,4 +450,46 @@ struct ConnectionTests {
         // verify connection hasnt been closed
         #expect(channel.isActive == true)
     }
+
+    @Test
+    @available(valkeySwift 1.0, *)
+    func testCloseOnServeClose() async throws {
+        let channel = try await ServerBootstrap(group: NIOSingletons.posixEventLoopGroup)
+            .serverChannelOption(.socketOption(.so_reuseaddr), value: 1)
+            .childChannelOption(ChannelOptions.allowRemoteHalfClosure, value: true)
+            .childChannelInitializer { channel in
+                channel.eventLoop.makeCompletedFuture {
+                    try channel.pipeline.syncOperations.addHandler(
+                        TestValkeyServerChannelHandler { command, _, write in
+                            switch command {
+                            case "QUIT":
+                                write(ByteBuffer(string: "+2OK\r\n"))
+                                channel.close(mode: .output, promise: nil)
+                            default:
+                                fatalError("Unexpected command: \(command)")
+                            }
+
+                        }
+                    )
+                }
+            }
+            .bind(host: "127.0.0.1", port: 0)
+            .get()
+        let port = channel.localAddress!.port!
+        try await ValkeyConnection.withConnection(
+            address: .hostname("127.0.0.1", port: port),
+            configuration: .init(),
+            eventLoop: MultiThreadedEventLoopGroup.singleton.any(),
+            logger: Logger(label: "test")
+        ) { connection in
+            let clientChannel = await connection.channel
+            try await connection.quit()
+            await withCheckedContinuation { cont in
+                clientChannel.closeFuture.whenComplete { _ in
+                    cont.resume()
+                }
+            }
+        }
+        try await channel.close()
+    }
 }