From 8cdcffd8cdf0ec3bf0781a2b40d46669ed226fa3 Mon Sep 17 00:00:00 2001 From: Nathan Harris Date: Wed, 9 Jan 2019 14:09:17 -0800 Subject: [PATCH] Add `RedisCommandHandler` to eventually replace `RedisMessenger` that works more deeply with NIO --- .../ChannelHandlers/RedisCommandHandler.swift | 78 +++++++++++++++++++ .../ChannelHandlers/RedisMessenger.swift | 1 + 2 files changed, 79 insertions(+) create mode 100644 Sources/NIORedis/ChannelHandlers/RedisCommandHandler.swift diff --git a/Sources/NIORedis/ChannelHandlers/RedisCommandHandler.swift b/Sources/NIORedis/ChannelHandlers/RedisCommandHandler.swift new file mode 100644 index 0000000..ab6cfee --- /dev/null +++ b/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 +} + +/// 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] + + 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?) { + let context = unwrapOutboundIn(data) + commandResponseQueue.insert(context.promise, at: 0) + ctx.write(wrapOutboundOut(context.command), promise: promise) + } +} diff --git a/Sources/NIORedis/ChannelHandlers/RedisMessenger.swift b/Sources/NIORedis/ChannelHandlers/RedisMessenger.swift index 52ff673..3e8130a 100644 --- a/Sources/NIORedis/ChannelHandlers/RedisMessenger.swift +++ b/Sources/NIORedis/ChannelHandlers/RedisMessenger.swift @@ -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