Merge branch '47-proposal-feedback' into 'master'

47 -- SSWG Review Feedback

Closes #55, #57, #54, #49, and #47

See merge request Mordil/swift-redis-nio-client!53

Author: Mordil

Package.swift

@@ -27,6 +27,7 @@ let package = Package(
     ],
     targets: [
         .target(name: "RedisNIO", dependencies: ["NIO", "Logging", "Metrics"]),
-        .testTarget(name: "RedisNIOTests", dependencies: ["RedisNIO", "NIO"])
+        .target(name: "RedisNIOTestUtils", dependencies: ["NIO", "RedisNIO"]),
+        .testTarget(name: "RedisNIOTests", dependencies: ["RedisNIO", "NIO", "RedisNIOTestUtils"])
     ]
 )

README.md

@@ -24,26 +24,28 @@ To install **RedisNIO**, just add the package as a dependency in your [**Package
 
 ```swift
 dependencies: [
-    .package(url: "https://github.com/Mordil/swift-redis-nio-client.git", from: "1.0.0-alpha.1")
+    .package(url: "https://github.com/Mordil/swift-redis-nio-client.git", from: "1.0.0-alpha.4")
 ]
 ```
 
 and run the following command: `swift package resolve`
 
 ## :zap: Getting Started
 
-**RedisNIO** is ready to use right after installation.
+**RedisNIO** is quick to use - all you need is an [`EventLoop`](https://apple.github.io/swift-nio/docs/current/NIO/Protocols/EventLoop.html) from **SwiftNIO**.
 
 ```swift
+import NIO
 import RedisNIO
 
-let connection = Redis.makeConnection(
+let eventLoop: EventLoop = ...
+let connection = RedisConnection.connect(
     to: try .init(ipAddress: "127.0.0.1", port: RedisConnection.defaultPort),
-    password: "my_pass"
+    on: eventLoop
 ).wait()
 
 let result = try connection.set("my_key", to: "some value")
-    .flatMap { return connection.get("my_key" }
+    .flatMap { return connection.get("my_key") }
     .wait()
 
 print(result) // Optional("some value")

Sources/RedisNIO/ChannelHandlers/RedisByteDecoder.swift

@@ -21,18 +21,19 @@ import NIO
 public final class RedisByteDecoder: ByteToMessageDecoder {
     /// `ByteToMessageDecoder.InboundOut`
     public typealias InboundOut = RESPValue
+    
+    private let parser: RESPTranslator
+    
+    public init() {
+        self.parser = RESPTranslator()
+    }
 
     /// See `ByteToMessageDecoder.decode(context:buffer:)`
     public func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
-        var position = 0
-
-        switch try RESPTranslator.parseBytes(&buffer, fromIndex: &position) {
-        case .incomplete: return .needMoreData
-        case let .parsed(value):
-            context.fireChannelRead(wrapInboundOut(value))
-            buffer.moveReaderIndex(forwardBy: position)
-            return .continue
-        }
+        guard let value = try self.parser.parseBytes(from: &buffer) else { return .needMoreData }
+        
+        context.fireChannelRead(wrapInboundOut(value))
+        return .continue
     }
 
     /// See `ByteToMessageDecoder.decodeLast(context:buffer:seenEOF)`

Sources/RedisNIO/ChannelHandlers/RedisCommandHandler.swift

@@ -16,11 +16,15 @@ import struct Foundation.UUID
 import Logging
 import NIO
 
-/// A context for `RedisCommandHandler` to operate within.
-public struct RedisCommandContext {
-    /// A full command keyword and arguments stored as a single `RESPValue`.
+/// The `NIO.ChannelOutboundHandler.OutboundIn` type for `RedisCommandHandler`.
+///
+/// This holds the command and its arguments stored as a single `RESPValue` to be sent to Redis,
+/// and an `NIO.EventLoopPromise` to be fulfilled when a response has been received.
+/// - Important: This struct has _reference semantics_ due to the retention of the `NIO.EventLoopPromise`.
+public struct RedisCommand {
+    /// A command keyword and its arguments stored as a single `RESPValue.array`.
     public let command: RESPValue
-    /// A promise expected to be fulfilled with the `RESPValue` response to the command from Redis.
+    /// A promise to be fulfilled with the sent command's response from Redis.
     public let responsePromise: EventLoopPromise<RESPValue>
 
     public init(command: RESPValue, promise: EventLoopPromise<RESPValue>) {
@@ -29,18 +33,28 @@ public struct RedisCommandContext {
     }
 }
 
-/// A `ChannelDuplexHandler` that works with `RedisCommandContext`s to send commands and forward responses.
-open class RedisCommandHandler {
-    /// Queue of promises waiting to receive a response value from a sent command.
+/// An object that operates in a First In, First Out (FIFO) request-response cycle.
+///
+/// `RedisCommandHandler` is a `NIO.ChannelDuplexHandler` that sends `RedisCommand` instances to Redis,
+/// and fulfills the command's `NIO.EventLoopPromise` as soon as a `RESPValue` response has been received from Redis.
+public final class RedisCommandHandler {
+    /// FIFO queue of promises waiting to receive a response value from a sent command.
     private var commandResponseQueue: CircularBuffer<EventLoopPromise<RESPValue>>
     private var logger: Logger
 
     deinit {
-        guard commandResponseQueue.count > 0 else { return }
-        logger.warning("Command handler deinit when queue is not empty. Current size: \(commandResponseQueue.count)")
+        guard self.commandResponseQueue.count > 0 else { return }
+        self.logger[metadataKey: "Queue Size"] = "\(self.commandResponseQueue.count)"
+        self.logger.warning("Command handler deinit when queue is not empty")
     }
 
-    public init(logger: Logger = Logger(label: "RedisNIO.CommandHandler"), initialQueueCapacity: Int = 5) {
+    /// - Parameters:
+    ///     - initialQueueCapacity: The initial queue size to start with. The default is `3`. `RedisCommandHandler` stores all
+    ///         `RedisCommand.responsePromise` objects into a buffer, and unless you intend to execute several concurrent commands against Redis,
+    ///         and don't want the buffer to resize, you shouldn't need to set this parameter.
+    ///     - logger: The `Logging.Logger` instance to use.
+    ///         The logger will have a `Foundation.UUID` value attached as metadata to uniquely identify this instance.
+    public init(initialQueueCapacity: Int = 3, logger: Logger = Logger(label: "RedisNIO.CommandHandler")) {
         self.commandResponseQueue = CircularBuffer(initialCapacity: initialQueueCapacity)
         self.logger = logger
         self.logger[metadataKey: "CommandHandler"] = "\(UUID())"
@@ -50,13 +64,16 @@ open class RedisCommandHandler {
 // MARK: ChannelInboundHandler
 
 extension RedisCommandHandler: ChannelInboundHandler {
-    /// See `ChannelInboundHandler.InboundIn`
+    /// See `NIO.ChannelInboundHandler.InboundIn`
     public typealias InboundIn = RESPValue
 
-    /// Invoked by NIO when an error has been thrown. The command response promise at the front of the queue will be
-    /// failed with the error.
+    /// Invoked by SwiftNIO when an error has been thrown. The command queue will be drained, with each promise in the queue being failed with the error thrown.
     ///
-    /// See `ChannelInboundHandler.errorCaught(context:error:)`
+    /// See `NIO.ChannelInboundHandler.errorCaught(context:error:)`
+    /// - Important: This will also close the socket connection to Redis.
+    /// - Note:`RedisMetrics.commandFailureCount` is **not** incremented from this error.
+    ///
+    /// A `Logging.LogLevel.critical` message will be written with the caught error.
     public func errorCaught(context: ChannelHandlerContext, error: Error) {
         let queue = self.commandResponseQueue
 
@@ -65,21 +82,22 @@ extension RedisCommandHandler: ChannelInboundHandler {
         self.commandResponseQueue.removeAll()
         queue.forEach { $0.fail(error) }
 
-        logger.critical("Error in channel pipeline.", metadata: ["error": .string(error.localizedDescription)])
-        
-        context.fireErrorCaught(error)
+        self.logger.critical("Error in channel pipeline.", metadata: ["error": "\(error.localizedDescription)"])
+
+        context.close(promise: nil)
     }
 
-    /// Invoked by NIO when a read has been fired from earlier in the response chain. This forwards the unwrapped
-    /// `RESPValue` to the promise awaiting a response at the front of the queue.
+    /// Invoked by SwiftNIO when a read has been fired from earlier in the response chain.
+    /// This forwards the decoded `RESPValue` response message to the promise waiting to be fulfilled at the front of the command queue.
+    /// - Note: `RedisMetrics.commandFailureCount` and `RedisMetrics.commandSuccessCount` are incremented from this method.
     ///
-    /// See `ChannelInboundHandler.channelRead(context:data:)`
+    /// See `NIO.ChannelInboundHandler.channelRead(context:data:)`
     public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
         let value = self.unwrapInboundIn(data)
 
         guard let leadPromise = self.commandResponseQueue.popFirst() else {
             assertionFailure("Read triggered with an empty promise queue! Ignoring: \(value)")
-            logger.critical("Read triggered with no promise waiting in the queue!")
+            self.logger.critical("Read triggered with no promise waiting in the queue!")
             return
         }
 
@@ -98,16 +116,16 @@ extension RedisCommandHandler: ChannelInboundHandler {
 // MARK: ChannelOutboundHandler
 
 extension RedisCommandHandler: ChannelOutboundHandler {
-    /// See `ChannelOutboundHandler.OutboundIn`
-    public typealias OutboundIn = RedisCommandContext
-    /// See `ChannelOutboundHandler.OutboundOut`
+    /// See `NIO.ChannelOutboundHandler.OutboundIn`
+    public typealias OutboundIn = RedisCommand
+    /// See `NIO.ChannelOutboundHandler.OutboundOut`
     public typealias OutboundOut = RESPValue
 
-    /// Invoked by NIO when a `write` has been requested on the `Channel`.
-    /// This unwraps a `RedisCommandContext`, retaining a callback to forward a response to later, and forwards
-    /// the underlying command data further into the pipeline.
+    /// Invoked by SwiftNIO when a `write` has been requested on the `Channel`.
+    /// This unwraps a `RedisCommand`, storing the `NIO.EventLoopPromise` in a command queue,
+    /// to fulfill later with the response to the command that is about to be sent through the `NIO.Channel`.
     ///
-    /// See `ChannelOutboundHandler.write(context:data:promise:)`
+    /// See `NIO.ChannelOutboundHandler.write(context:data:promise:)`
     public func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
         let commandContext = self.unwrapOutboundIn(data)
         self.commandResponseQueue.append(commandContext.responsePromise)

Sources/RedisNIO/ChannelHandlers/RedisMessageEncoder.swift

@@ -33,7 +33,7 @@ public final class RedisMessageEncoder: MessageToByteEncoder {
     ///
     /// See [https://redis.io/topics/protocol](https://redis.io/topics/protocol) and `RESPEncoder.encode(data:out:)`
     public func encode(data: RESPValue, out: inout ByteBuffer) throws {
-        RESPTranslator.writeValue(data, into: &out)
+        out.writeRESPValue(data)
 
         logger.debug("Encoded \(data) to \(getPrintableString(for: &out))")
     }

Sources/RedisNIO/Commands/BasicCommands.swift

@@ -22,7 +22,8 @@ extension RedisClient {
     /// - Returns: The message sent with the command.
     @inlinable
     public func echo(_ message: String) -> EventLoopFuture<String> {
-        return send(command: "ECHO", with: [message])
+        let args = [RESPValue(bulk: message)]
+        return send(command: "ECHO", with: args)
             .convertFromRESPValue()
     }
 
@@ -33,8 +34,10 @@ extension RedisClient {
     /// - Returns: The provided message or Redis' default response of `"PONG"`.
     @inlinable
     public func ping(with message: String? = nil) -> EventLoopFuture<String> {
-        let arg = message != nil ? [message] : []
-        return send(command: "PING", with: arg)
+        let args: [RESPValue] = message != nil
+            ? [.init(bulk: message!)] // safe because we did a nil pre-check
+            : []
+        return send(command: "PING", with: args)
             .convertFromRESPValue()
     }
 
@@ -46,7 +49,8 @@ extension RedisClient {
     /// - Returns: An `EventLoopFuture` that resolves when the operation has succeeded, or fails with a `RedisError`.
     @inlinable
     public func select(database index: Int) -> EventLoopFuture<Void> {
-        return send(command: "SELECT", with: [index])
+        let args = [RESPValue(bulk: index)]
+        return send(command: "SELECT", with: args)
             .map { _ in return () }
     }
 
@@ -59,8 +63,11 @@ extension RedisClient {
     /// - Returns: `true` if the swap was successful.
     @inlinable
     public func swapDatabase(_ first: Int, with second: Int) -> EventLoopFuture<Bool> {
-        /// connection.swapDatabase(index: 0, withIndex: 10)
-        return send(command: "SWAPDB", with: [first, second])
+        let args: [RESPValue] = [
+            .init(bulk: first),
+            .init(bulk: second)
+        ]
+        return send(command: "SWAPDB", with: args)
             .convertFromRESPValue(to: String.self)
             .map { return $0 == "OK" }
     }
@@ -74,7 +81,8 @@ extension RedisClient {
     public func delete(_ keys: [String]) -> EventLoopFuture<Int> {
         guard keys.count > 0 else { return self.eventLoop.makeSucceededFuture(0) }
 
-        return send(command: "DEL", with: keys)
+        let args = keys.map(RESPValue.init)
+        return send(command: "DEL", with: args)
             .convertFromRESPValue()
     }
 
@@ -84,12 +92,16 @@ extension RedisClient {
     /// [https://redis.io/commands/expire](https://redis.io/commands/expire)
     /// - Parameters:
     ///     - key: The key to set the expiration on.
-    ///     - deadline: The time from now the key will expire at.
+    ///     - timeout: The time from now the key will expire at.
     /// - Returns: `true` if the expiration was set.
     @inlinable
-    public func expire(_ key: String, after deadline: TimeAmount) -> EventLoopFuture<Bool> {
-        let amount = deadline.nanoseconds / 1_000_000_000
-        return send(command: "EXPIRE", with: [key, amount])
+    public func expire(_ key: String, after timeout: TimeAmount) -> EventLoopFuture<Bool> {
+        let amount = timeout.nanoseconds / 1_000_000_000
+        let args: [RESPValue] = [
+            .init(bulk: key),
+            .init(bulk: amount.description)
+        ]
+        return send(command: "EXPIRE", with: args)
             .convertFromRESPValue(to: Int.self)
             .map { return $0 == 1 }
     }
@@ -116,7 +128,7 @@ extension RedisClient {
     }
 
     @usableFromInline
-    func _scan<T>(
+    internal func _scan<T>(
         command: String,
         resultType: T.Type = T.self,
         _ key: String?,
@@ -127,19 +139,19 @@ extension RedisClient {
         where
         T: RESPValueConvertible
     {
-        var args: [RESPValueConvertible] = [pos]
+        var args: [RESPValue] = [.init(bulk: pos)]
 
         if let k = key {
-            args.insert(k, at: 0)
+            args.insert(.init(bulk: k), at: 0)
         }
 
         if let m = match {
-            args.append("match")
-            args.append(m)
+            args.append(.init(bulk: "match"))
+            args.append(.init(bulk: m))
         }
         if let c = count {
-            args.append("count")
-            args.append(c)
+            args.append(.init(bulk: "count"))
+            args.append(.init(bulk: c))
         }
 
         let response = send(command: command, with: args).convertFromRESPValue(to: [RESPValue].self)