Command types (#5)

* Command types

* fix hello 3 command

* writeToRESPBuffer -> encode

* Add support for returning maps

Makes the assumption the map key is always a string

* Enum namespaces for containers

* Catch more response arrays in replies

* Parameter pack return from pipeline

* Update README

* Make sure we parse all responses from pipelined requests

Author: adam-fowler

README.md

@@ -51,14 +51,8 @@ In general you don't need to use this method as it has no advantages over using
 
 ```swift
 let key = RedisKey(rawValue: "MyKey")
-let responses = try await connection.pipeline([
-    .set(key, "TestString"),
-    .get(key)
-])
+let (setResponse, getResponse) = try await connection.pipeline(
+    SET(key, "TestString"),
+    GET(key)
+)
 ```
-
-The `RedisConnection.pipeline` command returns an array of `RESPTokens`, one for each command. So we can get the result of the `get` in the above example by converting the second token into a `String`.
-
-```swift
-let value = responses[1].converting(to: String.self)
-```

Sources/Redis/Connection/RedisConnection.swift

@@ -44,12 +44,12 @@ public struct ServerAddress: Sendable, Equatable {
 @_documentation(visibility: internal)
 public struct RedisConnection: Sendable {
     enum Request {
-        case command(RESPCommand)
-        case pipelinedCommands([RESPCommand])
+        case command(ByteBuffer)
+        case pipelinedCommands(ByteBuffer, Int)
     }
     enum Response {
         case token(RESPToken)
-        case pipelinedResponse([RESPToken])
+        case pipelinedResponse([Result<RESPToken, Error>])
     }
     typealias RequestStreamElement = (Request, CheckedContinuation<Response, Error>)
     /// Logger used by Server
@@ -96,7 +96,7 @@ public struct RedisConnection: Sendable {
                         do {
                             switch request {
                             case .command(let command):
-                                try await outbound.write(command.buffer)
+                                try await outbound.write(command)
                                 let response = try await inboundIterator.next()
                                 if let response {
                                     continuation.resume(returning: .token(response))
@@ -109,22 +109,26 @@ public struct RedisConnection: Sendable {
                                         )
                                     )
                                 }
-                            case .pipelinedCommands(let commands):
-                                try await outbound.write(contentsOf: commands.map { $0.buffer })
-                                var responses: [RESPToken] = .init()
-                                for _ in 0..<commands.count {
-                                    let response = try await inboundIterator.next()
-                                    if let response {
-                                        responses.append(response)
-                                    } else {
-                                        requestContinuation.finish()
-                                        continuation.resume(
-                                            throwing: RedisClientError(
-                                                .connectionClosed,
-                                                message: "The connection to the Redis database was unexpectedly closed."
+                            case .pipelinedCommands(let commands, let count):
+                                try await outbound.write(commands)
+                                var responses: [Result<RESPToken, Error>] = .init()
+                                for _ in 0..<count {
+                                    do {
+                                        let response = try await inboundIterator.next()
+                                        if let response {
+                                            responses.append(.success(response))
+                                        } else {
+                                            requestContinuation.finish()
+                                            continuation.resume(
+                                                throwing: RedisClientError(
+                                                    .connectionClosed,
+                                                    message: "The connection to the Redis database was unexpectedly closed."
+                                                )
                                             )
-                                        )
-                                        return
+                                            return
+                                        }
+                                    } catch {
+                                        responses.append(.failure(error))
                                     }
                                 }
                                 continuation.resume(returning: .pipelinedResponse(responses))
@@ -153,14 +157,11 @@ public struct RedisConnection: Sendable {
         }
     }
 
-    @discardableResult public func send(command: RESPCommand) async throws -> RESPToken {
-        if logger.logLevel <= .debug {
-            var buffer = command.buffer
-            let sending = try [String](from: RESPToken(consuming: &buffer)!).joined(separator: " ")
-            self.logger.debug("send: \(sending)")
-        }
+    @discardableResult public func send<Command: RedisCommand>(command: Command) async throws -> Command.Response {
+        var encoder = RedisCommandEncoder()
+        command.encode(into: &encoder)
         let response: Response = try await withCheckedThrowingContinuation { continuation in
-            switch requestContinuation.yield((.command(command), continuation)) {
+            switch requestContinuation.yield((.command(encoder.buffer), continuation)) {
             case .enqueued:
                 break
             case .dropped, .terminated:
@@ -175,12 +176,21 @@ public struct RedisConnection: Sendable {
             }
         }
         guard case .token(let token) = response else { preconditionFailure("Expected a single response") }
-        return token
+        return try .init(from: token)
     }
 
-    @discardableResult public func pipeline(_ commands: [RESPCommand]) async throws -> [RESPToken] {
+    @discardableResult public func pipeline<each Command: RedisCommand>(
+        _ commands: repeat each Command
+    ) async throws -> (repeat (each Command).Response) {
+        var count = 0
+        var encoder = RedisCommandEncoder()
+        for command in repeat each commands {
+            command.encode(into: &encoder)
+            count += 1
+        }
+
         let response: Response = try await withCheckedThrowingContinuation { continuation in
-            switch requestContinuation.yield((.pipelinedCommands(commands), continuation)) {
+            switch requestContinuation.yield((.pipelinedCommands(encoder.buffer, count), continuation)) {
             case .enqueued:
                 break
             case .dropped, .terminated:
@@ -195,21 +205,19 @@ public struct RedisConnection: Sendable {
             }
         }
         guard case .pipelinedResponse(let tokens) = response else { preconditionFailure("Expected a single response") }
-        return tokens
-    }
 
-    @discardableResult public func send<each Arg: RESPRenderable>(_ command: repeat each Arg) async throws -> RESPToken {
-        let command = RESPCommand(repeat each command)
-        return try await self.send(command: command)
+        var index = AutoIncrementingInteger()
+        return try (repeat (each Command).Response(from: tokens[index.next()].get()))
     }
 
     /// Try to upgrade to RESP3
     private func resp3Upgrade(
         outbound: NIOAsyncChannelOutboundWriter<ByteBuffer>,
         inboundIterator: inout NIOAsyncChannelInboundStream<RESPToken>.AsyncIterator
     ) async throws {
-        let helloCommand = RESPCommand("HELLO", "3")
-        try await outbound.write(helloCommand.buffer)
+        var encoder = RedisCommandEncoder()
+        encoder.encodeArray("HELLO", 3)
+        try await outbound.write(encoder.buffer)
         let response = try await inboundIterator.next()
         guard let response else {
             throw RedisClientError(.connectionClosed, message: "The connection to the Redis database was unexpectedly closed.")
@@ -317,3 +325,11 @@ extension ClientBootstrap: ClientBootstrapProtocol {}
 #if canImport(Network)
 extension NIOTSConnectionBootstrap: ClientBootstrapProtocol {}
 #endif
+
+private struct AutoIncrementingInteger {
+    var value: Int = 0
+    mutating func next() -> Int {
+        value += 1
+        return value - 1
+    }
+}

Sources/Redis/RESP/RESPError.swift

@@ -74,6 +74,7 @@ public struct RESPParsingError: Error {
     public var code: Code
     public var buffer: ByteBuffer
 
+    @usableFromInline
     package init(code: Code, buffer: ByteBuffer) {
         self.code = code
         self.buffer = buffer

Sources/Redis/RESP/RESPRenderable.swift

@@ -16,15 +16,15 @@ import NIOCore
 
 /// Type that can be rendered into a RESP buffer
 public protocol RESPRenderable {
-    func writeToRESPBuffer(_ buffer: inout ByteBuffer) -> Int
+    func encode(into commandEncoder: inout RedisCommandEncoder) -> Int
 }
 
 extension Optional: RESPRenderable where Wrapped: RESPRenderable {
     @inlinable
-    public func writeToRESPBuffer(_ buffer: inout ByteBuffer) -> Int {
+    public func encode(into commandEncoder: inout RedisCommandEncoder) -> Int {
         switch self {
         case .some(let wrapped):
-            return wrapped.writeToRESPBuffer(&buffer)
+            return wrapped.encode(into: &commandEncoder)
         case .none:
             return 0
         }
@@ -33,49 +33,35 @@ extension Optional: RESPRenderable where Wrapped: RESPRenderable {
 
 extension Array: RESPRenderable where Element: RESPRenderable {
     @inlinable
-    public func writeToRESPBuffer(_ buffer: inout ByteBuffer) -> Int {
+    public func encode(into commandEncoder: inout RedisCommandEncoder) -> Int {
         var count = 0
         for element in self {
-            count += element.writeToRESPBuffer(&buffer)
+            count += element.encode(into: &commandEncoder)
         }
         return count
     }
 }
 
 extension String: RESPRenderable {
     @inlinable
-    public func writeToRESPBuffer(_ buffer: inout ByteBuffer) -> Int {
-        buffer.writeBulkString(self)
+    public func encode(into commandEncoder: inout RedisCommandEncoder) -> Int {
+        commandEncoder.encodeBulkString(self)
         return 1
     }
 }
 
 extension Int: RESPRenderable {
     @inlinable
-    public func writeToRESPBuffer(_ buffer: inout ByteBuffer) -> Int {
-        buffer.writeBulkString(String(self))
+    public func encode(into commandEncoder: inout RedisCommandEncoder) -> Int {
+        commandEncoder.encodeBulkString(String(self))
         return 1
     }
 }
 
 extension Double: RESPRenderable {
     @inlinable
-    public func writeToRESPBuffer(_ buffer: inout ByteBuffer) -> Int {
-        buffer.writeBulkString(String(self))
+    public func encode(into commandEncoder: inout RedisCommandEncoder) -> Int {
+        commandEncoder.encodeBulkString(String(self))
         return 1
     }
 }
-
-extension ByteBuffer {
-    public mutating func writeRESP3TypeIdentifier(_ identifier: RESPTypeIdentifier) {
-        self.writeInteger(identifier.rawValue)
-    }
-
-    public mutating func writeBulkString(_ string: String) {
-        self.writeRESP3TypeIdentifier(.bulkString)
-        self.writeString(String(string.utf8.count))
-        self.writeStaticString("\r\n")
-        self.writeString(string)
-        self.writeStaticString("\r\n")
-    }
-}

Sources/Redis/RESP/RESPToken.swift

@@ -93,6 +93,7 @@ public struct RESPToken: Hashable, Sendable {
         case push(Array)
     }
 
+    @usableFromInline
     package let base: ByteBuffer
 
     public var value: Value {

Sources/Redis/RESP/RESPTokenRepresentable.swift

@@ -24,10 +24,12 @@ extension RESPToken: RESPTokenRepresentable {
     /// - Parameter type: Type to convert to
     /// - Throws: RedisClientError.unexpectedType
     /// - Returns: Value
+    @inlinable
     public func converting<Value: RESPTokenRepresentable>(to type: Value.Type = Value.self) throws -> Value {
         try Value(from: self)
     }
 
+    @inlinable
     public init(from token: RESPToken) throws {
         self = token
     }
@@ -38,12 +40,14 @@ extension Array where Element == RESPToken {
     /// - Parameter type: Type to convert to
     /// - Throws: RedisClientError.unexpectedType
     /// - Returns: Array of Value
+    @inlinable
     public func converting<Value: RESPTokenRepresentable>(to type: [Value].Type = [Value].self) throws -> [Value] {
         try self.map { try $0.converting() }
     }
 }
 
 extension ByteBuffer: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .simpleString(let buffer), .bulkString(let buffer), .verbatimString(let buffer), .bigNumber(let buffer):
@@ -55,13 +59,15 @@ extension ByteBuffer: RESPTokenRepresentable {
 }
 
 extension String: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         let buffer = try ByteBuffer(from: token)
         self.init(buffer: buffer)
     }
 }
 
 extension Int: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .number(let value):
@@ -73,6 +79,7 @@ extension Int: RESPTokenRepresentable {
 }
 
 extension Double: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .double(let value):
@@ -84,6 +91,7 @@ extension Double: RESPTokenRepresentable {
 }
 
 extension Bool: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .boolean(let value):
@@ -95,6 +103,7 @@ extension Bool: RESPTokenRepresentable {
 }
 
 extension Optional: RESPTokenRepresentable where Wrapped: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .null:
@@ -106,6 +115,7 @@ extension Optional: RESPTokenRepresentable where Wrapped: RESPTokenRepresentable
 }
 
 extension Array: RESPTokenRepresentable where Element: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .array(let respArray), .push(let respArray):
@@ -122,6 +132,7 @@ extension Array: RESPTokenRepresentable where Element: RESPTokenRepresentable {
 }
 
 extension Set: RESPTokenRepresentable where Element: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .set(let respSet):
@@ -138,6 +149,7 @@ extension Set: RESPTokenRepresentable where Element: RESPTokenRepresentable {
 }
 
 extension Dictionary: RESPTokenRepresentable where Value: RESPTokenRepresentable, Key: RESPTokenRepresentable {
+    @inlinable
     public init(from token: RESPToken) throws {
         switch token.value {
         case .map(let respMap), .attribute(let respMap):

Sources/Redis/RedisCommand.swift

@@ -0,0 +1,22 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-redis open source project
+//
+// Copyright (c) 2025 the swift-redis project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-redis project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+
+/// A redis command that can be executed on a connection.
+public protocol RedisCommand {
+    associatedtype Response: RESPTokenRepresentable = RESPToken
+
+    func encode(into commandEncoder: inout RedisCommandEncoder)
+}