Merge branch 'pipelining'

* pipelining:
  Add `NIORedisPipeline` and `RedisPipeline` to allow chaining requests and receiving a single response
  Rename `redisPipeline` property in `NIORedisConnection` to remove possible confusion on upcoming `NIORedisPipeline` structure
  Restructure `RedisMessenger` to be a little more legible

Author: Mordil

Sources/NIORedis/Channel/RedisMessenger.swift

@@ -1,38 +1,81 @@
 import NIO
 
-/// `ChannelInboundHandler` that handles the responsibility of coordinating incoming and outgoing messages
-/// on a particular connection to Redis.
-internal final class RedisMessenger: ChannelInboundHandler {
-    /// See `ChannelInboundHandler.InboundIn`
-    public typealias InboundIn = RedisData
-
-    /// See `ChannelInboundHandler.OutboundOut`
-    public typealias OutboundOut = RedisData
-
-    /// Queue of promises waiting to receive an incoming response value from a outgoing message.
-    private var waitingResponseQueue: [EventLoopPromise<InboundIn>]
-    /// Queue of unsent outgoing messages, with the oldest objects at the end of the array.
-    private var outgoingMessageQueue: [OutboundOut]
-
-    /// This handler's event loop.
+/// `ChannelInboundHandler` that is responsible for coordinating incoming and outgoing messages on a particular
+/// connection to Redis.
+internal final class RedisMessenger {
     private let eventLoop: EventLoop
 
-    /// Context used for writing outgoing messages with.
-    private weak var outputContext: ChannelHandlerContext?
+    /// Context to be used for writing outgoing messages with.
+    private var channelContext: ChannelHandlerContext?
+
+    /// Queue of promises waiting to receive an incoming response value from an outgoing message.
+    private var waitingResponseQueue: [EventLoopPromise<RedisData>]
+    /// Queue of unset outgoing messages, with the oldest messages at the end of the array.
+    private var outgoingMessageQueue: [RedisData]
 
     /// Creates a new handler that works on the specified `EventLoop`.
-    public init(on eventLoop: EventLoop) {
+    init(on eventLoop: EventLoop) {
         self.waitingResponseQueue = []
         self.outgoingMessageQueue = []
         self.eventLoop = eventLoop
     }
+    
+    /// Adds a complete message encoded as `RedisData` to the queue and returns an `EventLoopFuture` that resolves
+    /// the response from Redis.
+    func enqueue(_ output: RedisData) -> EventLoopFuture<RedisData> {
+        // ensure that we are on the event loop before modifying our data
+        guard eventLoop.inEventLoop else {
+            return eventLoop.submit({}).then { return self.enqueue(output) }
+        }
+
+        // add the new output to the writing queue at the front
+        outgoingMessageQueue.insert(output, at: 0)
+
+        // every outgoing message is expected to receive some form of response, so create a promise that we'll resolve
+        // with the response
+        let promise = eventLoop.makePromise(of: RedisData.self)
+        waitingResponseQueue.insert(promise, at: 0)
+
+        // if we have a context for writing, flush the outgoing queue
+        channelContext?.eventLoop.execute {
+            self._flushOutgoingQueue()
+        }
+
+        return promise.futureResult
+    }
+
+    /// Writes all queued outgoing messages to the channel.
+    func _flushOutgoingQueue() {
+        guard let context = channelContext else { return }
+
+        while let output = outgoingMessageQueue.popLast() {
+            context.write(wrapOutboundOut(output), promise: nil)
+            context.flush()
+        }
+    }
+}
 
+// MARK: ChannelInboundHandler
+
+extension RedisMessenger: ChannelInboundHandler {
+    /// See `ChannelInboundHandler.InboundIn`
+    public typealias InboundIn = RedisData
+
+    /// See `ChannelInboundHandler.OutboundOut`
+    public typealias OutboundOut = RedisData
+
+    /// Invoked by NIO when the channel for this handler has become active, receiving a context that is ready to
+    /// send messages.
+    ///
+    /// Any queued messages will be flushed at this point.
     /// See `ChannelInboundHandler.channelActive(ctx:)`
     public func channelActive(ctx: ChannelHandlerContext) {
-        outputContext = ctx
+        channelContext = ctx
         _flushOutgoingQueue()
     }
 
+    /// Invoked by NIO when an error was thrown earlier in the response chain. The waiting 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 = waitingResponseQueue.last else {
@@ -41,6 +84,8 @@ internal final class RedisMessenger: ChannelInboundHandler {
         leadPromise.fail(error: error)
     }
 
+    /// Invoked by NIO when a read has been fired from earlier in the response chain. This forwards the unwrapped
+    /// `RedisData` to the response at the front of the queue.
     /// See `ChannelInboundHandler.channelRead(ctx:data:)`
     public func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
         let input = unwrapInboundIn(data)
@@ -54,38 +99,4 @@ internal final class RedisMessenger: ChannelInboundHandler {
 
         leadPromise.succeed(result: input)
     }
-
-    /// Adds a complete message encoded as `RedisData` to the queue and returns an `EventLoopFuture` that resolves
-    /// the response from Redis.
-    func enqueue(_ output: OutboundOut) -> EventLoopFuture<InboundIn> {
-        // ensure that we are on the event loop before modifying our data
-        guard eventLoop.inEventLoop else {
-            return eventLoop.submit { }.then { return self.enqueue(output) }
-        }
-
-        // add the new output to the writing queue at the front
-        outgoingMessageQueue.insert(output, at: 0)
-
-        // every outgoing message is expected to receive some form of response, so build
-        // the context in the readQueue that we resolve with the response
-        let promise = eventLoop.makePromise(of: InboundIn.self)
-        waitingResponseQueue.insert(promise, at: 0)
-
-        // if we have a context for writing, flush the outgoing queue
-        outputContext?.eventLoop.execute {
-            self._flushOutgoingQueue()
-        }
-
-        return promise.futureResult
-    }
-
-    /// Writes all queued outgoing messages to the channel.
-    func _flushOutgoingQueue() {
-        guard let context = outputContext else { return }
-
-        while let output = outgoingMessageQueue.popLast() {
-            context.write(wrapOutboundOut(output), promise: nil)
-            context.flush()
-        }
-    }
 }

Sources/NIORedis/NIORedisConnection.swift

@@ -8,7 +8,7 @@ public final class NIORedisConnection {
     /// Has the connection been closed?
     public private(set) var isClosed = Atomic<Bool>(value: false)
 
-    internal let redisPipeline: RedisMessenger
+    internal let messenger: RedisMessenger
 
     private let channel: Channel
 
@@ -18,7 +18,7 @@ public final class NIORedisConnection {
     /// - Important: Call `close()` before deinitializing to properly cleanup resources!
     init(channel: Channel, handler: RedisMessenger) {
         self.channel = channel
-        self.redisPipeline = handler
+        self.messenger = handler
     }
 
     /// Closes the connection to Redis.
@@ -31,7 +31,7 @@ public final class NIORedisConnection {
     /// Executes the desired command with the specified arguments.
     /// - Important: All arguments should be in `.bulkString` format.
     public func command(_ command: String, _ arguments: [RedisData] = []) -> EventLoopFuture<RedisData> {
-        return send(.array([RedisData(bulk: command)] + arguments))
+        return _send(.array([RedisData(bulk: command)] + arguments))
             .thenThrowing { response in
                 switch response {
                 case let .error(error): throw error
@@ -40,15 +40,20 @@ public final class NIORedisConnection {
             }
     }
 
-    private func send(_ message: RedisData) -> EventLoopFuture<RedisData> {
+    /// Creates a `NIORedisPipeline` for executing a batch of commands.
+    public func makePipeline() -> NIORedisPipeline {
+        return .init(using: self)
+    }
+
+    func _send(_ message: RedisData) -> EventLoopFuture<RedisData> {
         // ensure the connection is still open
         guard !isClosed.load() else { return eventLoop.makeFailedFuture(error: RedisError.connectionClosed) }
 
         // create a new promise to store
         let promise = eventLoop.makePromise(of: RedisData.self)
 
         // cascade this enqueue to the newly created promise
-        redisPipeline.enqueue(message).cascade(promise: promise)
+        messenger.enqueue(message).cascade(promise: promise)
 
         return promise.futureResult
     }

Sources/NIORedis/NIORedisPipeline.swift

@@ -0,0 +1,86 @@
+import Foundation
+import NIO
+
+/// An object that provides a mechanism to "pipeline" multiple Redis commands in sequence, providing an aggregate response
+/// of all the Redis responses for each individual command.
+///
+///     let results = connection.makePipeline()
+///         .enqueue(command: "SET", arguments: ["my_key", 3])
+///         .enqueue(command: "INCR", arguments: ["my_key"])
+///         .execute()
+///     // results == Future<[RedisData]>
+///     // results[0].string == Optional("OK")
+///     // results[1].int == Optional(4)
+/// - Important: The larger the pipeline queue, the more memory both NIORedis and Redis will use.
+/// See https://redis.io/topics/pipelining#redis-pipelining
+public final class NIORedisPipeline {
+    /// The client to execute the commands on.
+    private let connection: NIORedisConnection
+    private let encoder: RedisDataEncoder = .init()
+
+    /// The queue of complete, encoded commands to execute.
+    private var queue: [RedisData]
+    private var messageCount: Int
+
+    /// Creates a new pipeline queue using the provided `NIORedisConnection`.
+    /// - Parameter using: The connection to execute the commands on.
+    public init(using connection: NIORedisConnection) {
+        self.connection = connection
+        self.queue = []
+        self.messageCount = 0
+    }
+
+    /// Queues the provided command and arguments to be executed when `execute()` is invoked.
+    /// - Parameters:
+    ///     - command: The command to execute. See https://redis.io/commands
+    ///     - arguments: The arguments, if any, to send with the command.
+    /// - Returns: A self-reference to this `NIORedisPipeline` instance for chaining commands.
+    @discardableResult
+    public func enqueue(command: String, arguments: [RedisDataConvertible] = []) throws -> NIORedisPipeline {
+        let args = try arguments.map { try $0.convertToRedisData() }
+
+        queue.append(.array([RedisData(bulk: command)] + args))
+
+        return self
+    }
+
+    /// Flushes the queue, sending all of the commands to Redis in the same order as they were enqueued.
+    /// - Important: If any of the commands fail, the remaining commands will not execute and the `EventLoopFuture` will fail.
+    /// - Returns: A `EventLoopFuture` that resolves the `RedisData` responses, in the same order as the command queue.
+    public func execute() -> EventLoopFuture<[RedisData]> {
+        let promise = connection.eventLoop.makePromise(of: [RedisData].self)
+
+        var results = [RedisData]()
+        var iterator = queue.makeIterator()
+
+        // recursive internal method for chaining each request and
+        // attaching callbacks for failing or ultimately succeeding
+        func handle(_ command: RedisData) {
+            let future = connection._send(command)
+            future.whenSuccess { response in
+                switch response {
+                case let .error(error): promise.fail(error: error)
+                default:
+                    results.append(response)
+
+                    if let next = iterator.next() {
+                        handle(next)
+                    } else {
+                        promise.succeed(result: results)
+                    }
+                }
+            }
+            future.whenFailure { promise.fail(error: $0) }
+        }
+
+        if let first = iterator.next() {
+            handle(first)
+        } else {
+            promise.succeed(result: [])
+        }
+
+        promise.futureResult.whenComplete { self.queue = [] }
+
+        return promise.futureResult
+    }
+}

Sources/Redis/RedisConnection.swift

@@ -2,22 +2,28 @@ import Foundation
 import NIORedis
 
 public final class RedisConnection {
-    private let driverConnection: NIORedisConnection
+    let _driverConnection: NIORedisConnection
+
     private let queue: DispatchQueue
 
-    deinit { driverConnection.close() }
+    deinit { _driverConnection.close() }
 
     init(driver: NIORedisConnection, callbackQueue: DispatchQueue) {
-        self.driverConnection = driver
+        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)
+    }
+
     public func get(
         _ key: String,
         _ callback: @escaping (Result<String?, Error>
     ) -> Void) {
         // TODO: Make this a generic method to avoid copy/paste
-        driverConnection.get(key)
+        _driverConnection.get(key)
             .map { result in
                 self.queue.async { callback(.success(result)) }
             }

Sources/Redis/RedisPipeline.swift

@@ -0,0 +1,52 @@
+import Foundation
+import NIORedis
+
+/// An object that provides a mechanism to "pipeline" multiple Redis commands in sequence, providing an aggregate response
+/// of all the Redis responses for each individual command.
+///
+///     connection.makePipeline()
+///         .enqueue(command: "SET", arguments: ["my_key", 3])
+///         .enqueue(command: "INCR", arguments: ["my_key"])
+///         .execute { results in
+///             // results[0].string == Optional("OK")
+///             // results[1].int == Optional(4)
+///         }
+/// - 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 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)
+        self.queue = callbackQueue
+    }
+
+    /// Queues the provided command and arguments to be executed when `execute()` is invoked.
+    /// - Parameters:
+    ///     - command: The command to execute. See https://redis.io/commands
+    ///     - arguments: The arguments, if any, to send with the command.
+    /// - Returns: A self-reference to this `RedisPipeline` instance for chaining commands.
+    @discardableResult
+    public func enqueue(command: String, arguments: [RedisDataConvertible] = []) throws -> RedisPipeline {
+        try _driverPipeline.enqueue(command: command, arguments: arguments)
+        return self
+    }
+
+    /// Flushes the queue, sending all of the commands to Redis in the same order as they were enqueued.
+    /// - Important: If any of the commands fail, the remaining commands will not execute and the callback will receive a failure.
+    /// - Parameter callback: The callback to receive the results of the pipeline of commands, or an error if thrown.
+    public func execute(_ callback: @escaping (Result<[RedisData], Error>) -> Void) {
+        _driverPipeline.execute()
+            .map { results in
+                self.queue.async { callback(.success(results)) }
+            }
+            .whenFailure { error in
+                self.queue.async { callback(.failure(error)) }
+            }
+    }
+}