Merge pull request #11 from tanner0101/nio2-ctx-fixes

nio 2 ctx fixes

Author: Mordil

Sources/NIORedis/ChannelHandlers/RESPDecoder+ByteToMessage.swift

@@ -4,18 +4,23 @@ extension RESPDecoder: ByteToMessageDecoder {
     /// `ByteToMessageDecoder.InboundOut`
     public typealias InboundOut = RESPValue
 
-    /// See `ByteToMessageDecoder.decode(ctx:buffer:)`
-    public func decode(ctx: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
+    /// See `ByteToMessageDecoder.decode(context:buffer:)`
+    public func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
         var position = 0
 
         switch try parse(at: &position, from: &buffer) {
         case .notYetParsed:
             return .needMoreData
 
         case .parsed(let RESPValue):
-            ctx.fireChannelRead(wrapInboundOut(RESPValue))
+            context.fireChannelRead(wrapInboundOut(RESPValue))
             buffer.moveReaderIndex(forwardBy: position)
             return .continue
         }
     }
+    
+    /// See `ByteToMessageDecoder.decodeLast(context:buffer:seenEOF)`
+    public func decodeLast(context: ChannelHandlerContext, buffer: inout ByteBuffer, seenEOF: Bool) throws -> DecodingState {
+        return .needMoreData
+    }
 }

Sources/NIORedis/ChannelHandlers/RESPEncoder+MessageToByte.swift

@@ -4,8 +4,8 @@ extension RESPEncoder: MessageToByteEncoder {
     /// See `MessageToByteEncoder.OutboundIn`
     public typealias OutboundIn = RESPValue
 
-    /// See `RESPEncoder.encode(ctx:data:out:)`
-    public func encode(ctx: ChannelHandlerContext, data: RESPValue, out: inout ByteBuffer) throws {
+    /// See `RESPEncoder.encode(context:data:out:)`
+    public func encode(context: ChannelHandlerContext, data: RESPValue, out: inout ByteBuffer) throws {
         out.writeBytes(encode(data))
     }
 }

Sources/NIORedis/ChannelHandlers/RedisCommandHandler.swift

@@ -33,8 +33,8 @@ extension RedisCommandHandler: ChannelInboundHandler {
     /// 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) {
+    /// See `ChannelInboundHandler.errorCaught(context:error:)`
+    public func errorCaught(context: ChannelHandlerContext, error: Error) {
         guard let leadPromise = commandResponseQueue.last else {
             return assertionFailure("Received unexpected error while idle: \(error.localizedDescription)")
         }
@@ -44,8 +44,8 @@ extension RedisCommandHandler: ChannelInboundHandler {
     /// 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) {
+    /// See `ChannelInboundHandler.channelRead(context:data:)`
+    public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
         let value = unwrapInboundIn(data)
 
         guard let leadPromise = commandResponseQueue.last else {
@@ -74,10 +74,10 @@ extension RedisCommandHandler: ChannelOutboundHandler {
     /// 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.responsePromise, at: 0)
-        ctx.write(wrapOutboundOut(context.command), promise: promise)
+    /// See `ChannelOutboundHandler.write(context:data:promise:)`
+    public func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
+        let commandContext = unwrapOutboundIn(data)
+        commandResponseQueue.insert(commandContext.responsePromise, at: 0)
+        context.write(wrapOutboundOut(commandContext.command), promise: promise)
     }
 }

Sources/NIORedis/Extensions/NIO/ClientBootstrap.swift

@@ -17,7 +17,7 @@ extension ClientBootstrap {
                     ByteToMessageHandler(RESPDecoder()),
                     RedisCommandHandler()
                 ]
-                return .andAllSucceed(handlers.map { channel.pipeline.add(handler: $0) }, on: group.next())
+                return .andAllSucceed(handlers.map { channel.pipeline.addHandler($0) }, on: group.next())
             }
     }
 }

Sources/NIORedis/RedisDriver.swift

@@ -68,6 +68,6 @@ public final class RedisDriver {
 
 private extension ChannelPipeline {
     func addHandlers(_ handlers: ChannelHandler...) -> EventLoopFuture<Void> {
-        return .andAllSucceed(handlers.map { add(handler: $0) }, on: eventLoop)
+        return .andAllSucceed(handlers.map { addHandler($0) }, on: eventLoop)
     }
 }

Tests/NIORedisTests/ChannelHandlers/RESPDecoder+ByteToMessageDecoderTests.swift

@@ -58,12 +58,12 @@ final class RESPDecoderByteToMessageDecoderTests: XCTestCase {
         let embeddedChannel = EmbeddedChannel()
         defer { _ = try? embeddedChannel.finish() }
         let handler = ByteToMessageHandler(decoder)
-        try embeddedChannel.pipeline.add(handler: handler).wait()
+        try embeddedChannel.pipeline.addHandler(handler).wait()
         let context = try embeddedChannel.pipeline.context(handler: handler).wait()
 
         buffer.writeString(input)
 
-        return try decoder.decode(ctx: context, buffer: &buffer)
+        return try decoder.decode(context: context, buffer: &buffer)
     }
 }
 

Tests/NIORedisTests/ChannelHandlers/RESPDecoderTests.swift

@@ -110,7 +110,7 @@ final class RESPDecoderTests: XCTestCase {
         let embeddedChannel = EmbeddedChannel()
         defer { _ = try? embeddedChannel.finish() }
         let handler = ByteToMessageHandler(decoder)
-        try embeddedChannel.pipeline.add(handler: handler).wait()
+        try embeddedChannel.pipeline.addHandler(handler).wait()
         var buffer = allocator.buffer(capacity: 256)
         buffer.writeBytes(input)
         try embeddedChannel.writeInbound(buffer)
@@ -172,7 +172,7 @@ extension RESPDecoderTests {
         let embeddedChannel = EmbeddedChannel()
         defer { _ = try? embeddedChannel.finish() }
         let handler = ByteToMessageHandler(decoder)
-        try embeddedChannel.pipeline.add(handler: handler).wait()
+        try embeddedChannel.pipeline.addHandler(handler).wait()
 
         var buffer = allocator.buffer(capacity: 256)
         for message in AllData.messages {

Tests/NIORedisTests/ChannelHandlers/RESPEncoderTests.swift

@@ -13,7 +13,7 @@ final class RESPEncoderTests: XCTestCase {
         encoder = RESPEncoder()
         allocator = ByteBufferAllocator()
         channel = EmbeddedChannel()
-        _ = try? channel.pipeline.add(handler: encoder).wait()
+        _ = try? channel.pipeline.addHandler(encoder).wait()
     }
 
     override func tearDown() {
@@ -93,7 +93,7 @@ final class RESPEncoderTests: XCTestCase {
         let context = try channel.pipeline.context(handler: encoder).wait()
 
         var buffer = allocator.buffer(capacity: 256)
-        try encoder.encode(ctx: context, data: input, out: &buffer)
+        try encoder.encode(context: context, data: input, out: &buffer)
         validation(buffer)
     }
 }