Merge branch 'composeability' into rev-2

* composeability:
  Update unit test references for new type names and to cover more test cases
  Update `DispatchRedis` module with new classes
  Remove deprecated classes
  Update command extensions
  Add `RedisDriver` to eventually replace `NIORedis` that works more deeply with NIO
  Add `RedisPipeline` to eventually replace `NIORedisPipeline` that works more deeply with NIO
  Add `RedisConnection` to eventually replace `NIORedisConnection` that works more deeply with NIO
  Add `RedisCommandHandler` to eventually replace `RedisMessenger` that works more deeply with NIO

Author: Mordil

Sources/DispatchRedis/Redis.swift

@@ -4,12 +4,12 @@ import NIORedis
 
 /// A factory that handles all necessary details for creating `RedisConnection` instances.
 public final class Redis {
-    private let driver: NIORedis
+    private let driver: RedisDriver
 
     deinit { try? driver.terminate() }
 
     public init(threadCount: Int = 1) {
-        self.driver = NIORedis(executionModel: .spawnThreads(threadCount))
+        self.driver = RedisDriver(executionModel: .spawnThreads(threadCount))
     }
 
     public func makeConnection(

Sources/DispatchRedis/RedisConnection.swift

@@ -2,20 +2,20 @@ import Foundation
 import NIORedis
 
 public final class RedisConnection {
-    let _driverConnection: NIORedisConnection
+    let _driverConnection: NIORedis.RedisConnection
 
     private let queue: DispatchQueue
 
     deinit { _driverConnection.close() }
 
-    init(driver: NIORedisConnection, callbackQueue: DispatchQueue) {
+    init(driver: NIORedis.RedisConnection, callbackQueue: DispatchQueue) {
         self._driverConnection = driver
         self.queue = callbackQueue
     }
 
     /// Creates a `RedisPipeline` for executing a batch of commands.
     public func makePipeline(callbackQueue: DispatchQueue? = nil) -> RedisPipeline {
-        return .init(using: self, callbackQueue: callbackQueue ?? queue)
+        return .init(connection: self, callbackQueue: callbackQueue ?? queue)
     }
 
     public func get(

Sources/DispatchRedis/RedisPipeline.swift

@@ -14,15 +14,15 @@ import NIORedis
 /// - Important: The larger the pipeline queue, the more memory both the Redis driver and Redis server will use.
 /// See https://redis.io/topics/pipelining#redis-pipelining
 public final class RedisPipeline {
-    private let _driverPipeline: NIORedisPipeline
+    private let _driverPipeline: NIORedis.RedisPipeline
     private let queue: DispatchQueue
 
     /// Creates a new pipeline queue using the provided `RedisConnection`, executing callbacks on the provided `DispatchQueue`.
     /// - Parameters:
     ///     - using: The connection to execute the commands on.
     ///     - callbackQueue: The queue to execute all callbacks on.
-    public init(using connection: RedisConnection, callbackQueue: DispatchQueue) {
-        self._driverPipeline = NIORedisPipeline(using: connection._driverConnection)
+    public init(connection: RedisConnection, callbackQueue: DispatchQueue) {
+        self._driverPipeline = NIORedis.RedisPipeline(channel: connection._driverConnection.channel)
         self.queue = callbackQueue
     }
 

Sources/NIORedis/ChannelHandlers/RedisCommandHandler.swift

@@ -0,0 +1,78 @@
+import Foundation
+import NIO
+
+/// A context for `RedisCommandHandler` to operate within.
+public struct RedisCommandContext {
+    /// A full command keyword and arguments stored as a single `RESPValue`.
+    public let command: RESPValue
+    /// A promise expected to be fulfilled with the `RESPValue` response to the command from Redis.
+    public let promise: EventLoopPromise<RESPValue>
+}
+
+/// 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.
+    private var commandResponseQueue: [EventLoopPromise<RESPValue>]
+
+    public init() {
+        self.commandResponseQueue = []
+    }
+}
+
+// MARK: ChannelInboundHandler
+
+extension RedisCommandHandler: ChannelInboundHandler {
+    /// See `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.
+    ///
+    /// See `ChannelInboundHandler.errorCaught(ctx:error:)`
+    public func errorCaught(ctx: ChannelHandlerContext, error: Error) {
+        guard let leadPromise = commandResponseQueue.last else {
+            return assertionFailure("Received unexpected error while idle: \(error.localizedDescription)")
+        }
+        leadPromise.fail(error: error)
+    }
+
+    /// 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.
+    ///
+    /// See `ChannelInboundHandler.channelRead(ctx:data:)`
+    public func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
+        let value = unwrapInboundIn(data)
+
+        guard let leadPromise = commandResponseQueue.last else {
+            return assertionFailure("Read triggered with an empty promise queue! Ignoring: \(value)")
+        }
+
+        let popped = commandResponseQueue.popLast()
+        assert(popped != nil)
+
+        switch value {
+        case .error(let e): leadPromise.fail(error: e)
+        default: leadPromise.succeed(result: value)
+        }
+    }
+}
+
+// MARK: ChannelOutboundHandler
+
+extension RedisCommandHandler: ChannelOutboundHandler {
+    /// See `ChannelOutboundHandler.OutboundIn`
+    public typealias OutboundIn = RedisCommandContext
+    /// See `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.
+    ///
+    /// See `ChannelOutboundHandler.write(ctx:data:promise:)`
+    public func write(ctx: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
+        let context = unwrapOutboundIn(data)
+        commandResponseQueue.insert(context.promise, at: 0)
+        ctx.write(wrapOutboundOut(context.command), promise: promise)
+    }
+}

Sources/NIORedis/ChannelHandlers/RedisMessenger.swift

@@ -1,21 +1,22 @@
 import Foundation
 import NIO
 
-extension NIORedisConnection {
+extension RedisConnection {
     /// Select the Redis logical database having the specified zero-based numeric index.
     /// New connections always use the database 0.
     ///
     /// https://redis.io/commands/select
     public func select(_ id: Int) -> EventLoopFuture<Void> {
-        return command("SELECT", [RESPValue(bulk: id.description)])
+        return command("SELECT", arguments: [RESPValue(bulk: id.description)])
             .map { _ in return () }
     }
 
     /// Request for authentication in a password-protected Redis server.
     ///
     /// https://redis.io/commands/auth
     public func authorize(with password: String) -> EventLoopFuture<Void> {
-        return command("AUTH", [RESPValue(bulk: password)]).map { _ in return () }
+        return command("AUTH", arguments: [RESPValue(bulk: password)])
+            .map { _ in return () }
     }
 
     /// Removes the specified keys. A key is ignored if it does not exist.
@@ -24,7 +25,7 @@ extension NIORedisConnection {
     /// - Returns: A future number of keys that were removed.
     public func delete(_ keys: String...) -> EventLoopFuture<Int> {
         let keyArgs = keys.map { RESPValue(bulk: $0) }
-        return command("DEL", keyArgs)
+        return command("DEL", arguments: keyArgs)
             .thenThrowing { res in
                 guard let count = res.int else {
                     throw RedisError(identifier: "delete", reason: "Unexpected response: \(res)")
@@ -41,7 +42,7 @@ extension NIORedisConnection {
     ///     - after: The lifetime (in seconds) the key will expirate at.
     /// - Returns: A future bool indicating if the expiration was set or not.
     public func expire(_ key: String, after deadline: Int) -> EventLoopFuture<Bool> {
-        return command("EXPIRE", [RESPValue(bulk: key), RESPValue(bulk: deadline.description)])
+        return command("EXPIRE", arguments: [RESPValue(bulk: key), RESPValue(bulk: deadline.description)])
             .thenThrowing { res in
                 guard let value = res.int else {
                     throw RedisError(identifier: "expire", reason: "Unexpected response: \(res)")
@@ -56,7 +57,7 @@ extension NIORedisConnection {
     ///
     /// https://redis.io/commands/get
     public func get(_ key: String) -> EventLoopFuture<String?> {
-        return command("GET", [RESPValue(bulk: key)])
+        return command("GET", arguments: [RESPValue(bulk: key)])
             .map { return $0.string }
     }
 
@@ -66,7 +67,7 @@ extension NIORedisConnection {
     ///
     /// https://redis.io/commands/set
     public func set(_ key: String, to value: String) -> EventLoopFuture<Void> {
-        return command("SET", [RESPValue(bulk: key), RESPValue(bulk: value)])
+        return command("SET", arguments: [RESPValue(bulk: key), RESPValue(bulk: value)])
             .map { _ in return () }
     }
 }