Add RedisCommandHandler to eventually replace RedisMessenger that works more deeply with NIO

This commit is contained in:
Nathan Harris
2019-01-09 18:14:24 -08:00
parent 19cbadc506
commit 8cdcffd8cd
2 changed files with 79 additions and 0 deletions
@@ -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)
}
}
@@ -2,6 +2,7 @@ import NIO
/// `ChannelInboundHandler` that is responsible for coordinating incoming and outgoing messages on a particular
/// connection to Redis.
@available(*, deprecated)
internal final class RedisMessenger {
private let eventLoop: EventLoop