Backdeploy #100 fix and deprecate Channel Redis pipeline APIs

This is a backdeploy of commit cfb99ba.

This fixes issue #100

Author: Mordil

Sources/RediStack/Extensions/SwiftNIO.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the RediStack open source project
 //
-// Copyright (c) 2019-2020 RediStack project authors
+// Copyright (c) 2019-2022 RediStack project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information
@@ -26,7 +26,7 @@ extension TimeAmount {
 
 // MARK: Pipeline manipulation
 
-extension Channel {
+extension ChannelPipeline {
     /// Adds the baseline channel handlers needed to support sending and receiving messages in Redis Serialization Protocol (RESP) format to the pipeline.
     ///
     /// For implementation details, see `RedisMessageEncoder`, `RedisByteDecoder`, and `RedisCommandHandler`.
@@ -62,7 +62,7 @@ extension Channel {
             (RedisCommandHandler(), "RediStack.CommandHandler")
         ]
         return .andAllSucceed(
-            handlers.map { self.pipeline.addHandler($0, name: $1) },
+            handlers.map { self.addHandler($0, name: $1) },
             on: self.eventLoop
         )
     }
@@ -106,14 +106,38 @@ extension Channel {
     ///             | [ Socket.read ] |                | [ Socket.write ] |
     ///             +-----------------+                +------------------+
     /// - Returns: A `NIO.EventLoopFuture` that resolves the instance of the PubSubHandler that was added to the pipeline.
-    public func addPubSubHandler() -> EventLoopFuture<RedisPubSubHandler> {
-        return self.pipeline
-            .handler(type: RedisCommandHandler.self)
-            .flatMap {
-                let pubsubHandler = RedisPubSubHandler(eventLoop: self.eventLoop)
-                return self.pipeline
-                    .addHandler(pubsubHandler, name: "RediStack.PubSubHandler", position: .before($0))
-                    .map { pubsubHandler }
+    public func addRedisPubSubHandler() -> EventLoopFuture<RedisPubSubHandler> {
+        // first try to return the handler that already exists in the pipeline
+
+        return self.handler(type: RedisPubSubHandler.self)
+            .flatMapError {
+                // if it doesn't exist, add it to the pipeline
+                guard
+                    let error = $0 as? ChannelPipelineError,
+                    error == .notFound
+                else { return self.eventLoop.makeFailedFuture($0) }
+
+                return self.handler(type: RedisCommandHandler.self)
+                    .flatMap {
+                        let pubsubHandler = RedisPubSubHandler(eventLoop: self.eventLoop)
+                        return self.addHandler(pubsubHandler, name: "RediStack.PubSubHandler", position: .before($0))
+                            .map { pubsubHandler }
+                    }
+            }
+    }
+
+    /// Removes the provided Redis PubSub handler.
+    /// - Returns: A `NIO.EventLoopFuture` that resolves when the handler was removed from the pipeline.
+    public func removeRedisPubSubHandler(_ handler: RedisPubSubHandler) -> EventLoopFuture<Void> {
+        self.removeHandler(handler)
+            .flatMapError {
+                // if it was already removed, then we can just succeed
+                guard
+                    let error = $0 as? ChannelPipelineError,
+                    error == .alreadyRemoved
+                else { return self.eventLoop.makeFailedFuture($0) }
+
+                return self.eventLoop.makeSucceededVoidFuture()
             }
     }
 }
@@ -135,6 +159,6 @@ extension ClientBootstrap {
                 ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR),
                 value: 1
             )
-            .channelInitializer { $0.addBaseRedisHandlers() }
+            .channelInitializer { $0.pipeline.addBaseRedisHandlers() }
     }
 }

Sources/RediStack/RedisConnection.swift

@@ -451,8 +451,8 @@ extension RedisConnection {
         guard case let .pubsub(handler) = self.state else {
             logger.debug("not in pubsub mode, moving to pubsub mode")
             // otherwise, add it to the pipeline, add the subscriptions, and update our state after it was successful
-            return self.channel
-                .addPubSubHandler()
+            return self.channel.pipeline
+                .addRedisPubSubHandler()
                 .flatMap { handler in
                     logger.trace("handler added, adding subscription")
                     return handler
@@ -466,7 +466,8 @@ extension RedisConnection {
                             )
                             // if there was an error, no subscriptions were made
                             // so remove the handler and propogate the error to the caller by rethrowing it
-                            return self.channel.pipeline.removeHandler(handler)
+                            return self.channel.pipeline
+                                .removeRedisPubSubHandler(handler)
                                 .flatMapThrowing { throw error }
                         }
                         // success, return the handler
@@ -542,7 +543,8 @@ extension RedisConnection {
                 }
                 logger.debug("subscription removed, with no current active subscriptions. leaving pubsub mode")
                 // otherwise, remove the handler and update our state
-                return self.channel.pipeline.removeHandler(handler)
+                return self.channel.pipeline
+                    .removeRedisPubSubHandler(handler)
                     .map {
                         self.state = .open
                         logger.debug("connection is now open to all commands")

Sources/RediStack/_Deprecations.swift

@@ -139,3 +139,15 @@ extension RedisKey.Lifetime {
     @available(*, deprecated, renamed: "Duration")
     public typealias Lifetime = Duration
 }
+
+extension Channel {
+    @available(*, deprecated, renamed: "pipeline.addBaseRedisHandlers()")
+    public func addBaseRedisHandlers() -> EventLoopFuture<Void> {
+        return self.pipeline.addBaseRedisHandlers()
+    }
+
+    @available(*, deprecated, renamed: "pipeline.addRedisPubSubHandler()")
+    public func addPubSubHandler() -> EventLoopFuture<RedisPubSubHandler> {
+        return self.pipeline.addRedisPubSubHandler()
+    }
+}

Tests/RediStackIntegrationTests/Commands/PubSubCommandsTests.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the RediStack open source project
 //
-// Copyright (c) 2020 RediStack project authors
+// Copyright (c) 2020-2022 RediStack project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information
@@ -12,6 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+import NIO
 import RediStack
 import RediStackTestUtils
 import XCTest
@@ -340,3 +341,30 @@ final class RedisPubSubCommandsPoolTests: RediStackConnectionPoolIntegrationTest
         XCTAssertEqual(self.pool.leasedConnectionCount, 0)
     }
 }
+
+// MARK: - #100 subscribe race condition
+
+extension RedisPubSubCommandsTests {
+    func test_pubsub_pipelineChanges_hasNoRaceCondition() throws {
+        func runOperation(_ factory: (RedisChannelName) -> EventLoopFuture<Void>) -> EventLoopFuture<Void> {
+            return .andAllSucceed(
+                (0...100_000).reduce(into: []) {
+                    result, index in
+
+                    result.append(factory("\(#function)-\(index)"))
+                },
+                on: self.connection.eventLoop
+            )
+        }
+
+        // subscribing (adding handler)
+        try runOperation { self.connection.subscribe(to: $0) { _, _ in } }
+            .wait()
+
+        // unsubscribing (removing handler)
+        try runOperation { self.connection.unsubscribe(from: $0) }
+            .wait()
+
+        try self.connection.close().wait()
+    }
+}