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()
-        }
-    }
 }