[Redis] Add TLS support (#84)

joshuawright11 · web-flow · commit a6fc40c55c1c · 2022-03-31T14:27:22.000-07:00
Adds TLS support, pointing to a RediStack fork while we wait for the PR to be merged upstream.
diff --git a/Package.swift b/Package.swift
@@ -25,7 +25,7 @@ let package = Package(
         .package(url: "https://github.com/alchemy-swift/cron.git", from: "2.3.2"),
         .package(url: "https://github.com/alchemy-swift/pluralize", from: "1.0.1"),
         .package(url: "https://github.com/johnsundell/Plot.git", from: "0.8.0"),
-        .package(url: "https://github.com/Mordil/RediStack.git", from: "1.0.0"),
+        .package(url: "https://github.com/alchemy-swift/RediStack.git", branch: "ssl-support-1.2.0"),
         .package(url: "https://github.com/onevcat/Rainbow", .upToNextMajor(from: "4.0.0")),
         .package(url: "https://github.com/vadymmarkov/Fakery", from: "5.0.0"),
     ],
diff --git a/Sources/Alchemy/Redis/Redis+Commands.swift b/Sources/Alchemy/Redis/Redis+Commands.swift
@@ -1,62 +1,8 @@
 import NIO
 import RediStack
 
-/// RedisClient conformance. See `RedisClient` for docs.
-extension RedisClient: RediStack.RedisClient {
+extension RedisClient {
     
-    // MARK: RediStack.RedisClient
-    
-    public var eventLoop: EventLoop {
-        Loop.current
-    }
-    
-    public func logging(to logger: Logger) -> RediStack.RedisClient {
-        provider.getClient().logging(to: logger)
-    }
-    
-    public func send(command: String, with arguments: [RESPValue]) -> EventLoopFuture<RESPValue> {
-        provider.getClient()
-            .send(command: command, with: arguments).hop(to: Loop.current)
-    }
-    
-    public func subscribe(
-        to channels: [RedisChannelName],
-        messageReceiver receiver: @escaping RedisSubscriptionMessageReceiver,
-        onSubscribe subscribeHandler: RedisSubscriptionChangeHandler?,
-        onUnsubscribe unsubscribeHandler: RedisSubscriptionChangeHandler?
-    ) -> EventLoopFuture<Void> {
-        provider.getClient()
-            .subscribe(
-                to: channels,
-                messageReceiver: receiver,
-                onSubscribe: subscribeHandler,
-                onUnsubscribe: unsubscribeHandler
-            )
-    }
-
-    public func psubscribe(
-        to patterns: [String],
-        messageReceiver receiver: @escaping RedisSubscriptionMessageReceiver,
-        onSubscribe subscribeHandler: RedisSubscriptionChangeHandler?,
-        onUnsubscribe unsubscribeHandler: RedisSubscriptionChangeHandler?
-    ) -> EventLoopFuture<Void> {
-        provider.getClient()
-            .psubscribe(
-                to: patterns,
-                messageReceiver: receiver,
-                onSubscribe: subscribeHandler,
-                onUnsubscribe: unsubscribeHandler
-            )
-    }
-    
-    public func unsubscribe(from channels: [RedisChannelName]) -> EventLoopFuture<Void> {
-        provider.getClient().unsubscribe(from: channels)
-    }
-    
-    public func punsubscribe(from patterns: [String]) -> EventLoopFuture<Void> {
-        provider.getClient().punsubscribe(from: patterns)
-    }
-
     // MARK: - Alchemy sugar
 
     /// Wrapper around sending commands to Redis.
@@ -113,17 +59,3 @@ extension RedisClient: RediStack.RedisClient {
         }
     }
 }
-
-extension RedisConnection: RedisProvider {
-    public func getClient() -> RediStack.RedisClient {
-        self
-    }
-    
-    public func shutdown() throws {
-        try close().wait()
-    }
-    
-    public func transaction<T>(_ transaction: @escaping (RedisProvider) async throws -> T) async throws -> T {
-        try await transaction(self)
-    }
-}
diff --git a/Sources/Alchemy/Redis/RedisClient.swift b/Sources/Alchemy/Redis/RedisClient.swift
@@ -1,9 +1,10 @@
 import NIO
 import NIOConcurrencyHelpers
+import NIOSSL
 import RediStack
 
 /// A client for interfacing with a Redis instance.
-public struct RedisClient: Service {
+public struct RedisClient: Service, RediStack.RedisClient {
     public struct Identifier: ServiceIdentifier {
         private let hashable: AnyHashable
         public init(hashable: AnyHashable) { self.hashable = hashable }
@@ -20,15 +21,77 @@ public struct RedisClient: Service {
         try provider.shutdown()
     }
     
+    // MARK: RediStack.RedisClient
+    
+    public var eventLoop: EventLoop {
+        Loop.current
+    }
+    
+    public func logging(to logger: Logger) -> RediStack.RedisClient {
+        provider.logging(to: logger)
+    }
+    
+    public func send(command: String, with arguments: [RESPValue]) -> EventLoopFuture<RESPValue> {
+        wrapError {
+            try provider.getClient()
+                .send(command: command, with: arguments).hop(to: Loop.current)
+        }
+    }
+    
+    public func subscribe(
+        to channels: [RedisChannelName],
+        messageReceiver receiver: @escaping RedisSubscriptionMessageReceiver,
+        onSubscribe subscribeHandler: RedisSubscriptionChangeHandler?,
+        onUnsubscribe unsubscribeHandler: RedisSubscriptionChangeHandler?
+    ) -> EventLoopFuture<Void> {
+        wrapError {
+            try provider.getClient()
+                .subscribe(
+                    to: channels,
+                    messageReceiver: receiver,
+                    onSubscribe: subscribeHandler,
+                    onUnsubscribe: unsubscribeHandler
+                )
+        }
+    }
+
+    public func psubscribe(
+        to patterns: [String],
+        messageReceiver receiver: @escaping RedisSubscriptionMessageReceiver,
+        onSubscribe subscribeHandler: RedisSubscriptionChangeHandler?,
+        onUnsubscribe unsubscribeHandler: RedisSubscriptionChangeHandler?
+    ) -> EventLoopFuture<Void> {
+        wrapError {
+            try provider.getClient()
+                .psubscribe(
+                    to: patterns,
+                    messageReceiver: receiver,
+                    onSubscribe: subscribeHandler,
+                    onUnsubscribe: unsubscribeHandler
+                )
+        }
+    }
+    
+    public func unsubscribe(from channels: [RedisChannelName]) -> EventLoopFuture<Void> {
+        wrapError { try provider.getClient().unsubscribe(from: channels) }
+    }
+    
+    public func punsubscribe(from patterns: [String]) -> EventLoopFuture<Void> {
+        wrapError { try provider.getClient().punsubscribe(from: patterns) }
+    }
+    
+    // MARK: Creating
+    
     /// A single redis connection
     public static func connection(
         _ host: String,
         port: Int = 6379,
         password: String? = nil,
         database: Int? = nil,
-        poolSize: RedisConnectionPoolSize = .maximumActiveConnections(1)
+        poolSize: RedisConnectionPoolSize = .maximumActiveConnections(1),
+        tlsConfiguration: TLSConfiguration? = nil
     ) -> RedisClient {
-        return .cluster(.ip(host: host, port: port), password: password, database: database, poolSize: poolSize)
+        return .cluster(.ip(host: host, port: port), password: password, database: database, poolSize: poolSize, tlsConfiguration: tlsConfiguration)
     }
     
     /// Convenience initializer for creating a redis client with the
@@ -48,29 +111,21 @@ public struct RedisClient: Service {
         _ sockets: Socket...,
         password: String? = nil,
         database: Int? = nil,
-        poolSize: RedisConnectionPoolSize = .maximumActiveConnections(1)
+        poolSize: RedisConnectionPoolSize = .maximumActiveConnections(1),
+        tlsConfiguration: TLSConfiguration? = nil
     ) -> RedisClient {
         return .configuration(
             RedisConnectionPool.Configuration(
-                initialServerConnectionAddresses: sockets.map {
-                    do {
-                        switch $0 {
-                        case let .ip(host, port):
-                            return try .makeAddressResolvingHost(host, port: port)
-                        case let .unix(path):
-                            return try .init(unixDomainSocketPath: path)
-                        }
-                    } catch {
-                        fatalError("Error generating socket address from `Socket` \(self)!")
-                    }
-                },
+                initialServerConnectionAddresses: [],
                 maximumConnectionCount: poolSize,
                 connectionFactoryConfiguration: RedisConnectionPool.ConnectionFactoryConfiguration(
                     connectionInitialDatabase: database,
                     connectionPassword: password,
-                    connectionDefaultLogger: Log.logger
+                    connectionDefaultLogger: Log.logger,
+                    tlsConfiguration: tlsConfiguration
                 )
-            )
+            ),
+            addresses: sockets
         )
     }
     
@@ -83,44 +138,36 @@ public struct RedisClient: Service {
     public static func configuration(_ config: RedisConnectionPool.Configuration) -> RedisClient {
         return RedisClient(provider: ConnectionPool(config: config))
     }
-}
-
-/// Under the hood provider for `Redis`. Used so either connection pools
-/// or connections can be injected into `Redis` for accessing redis.
-public protocol RedisProvider {
-    /// Get a redis client for running commands.
-    func getClient() -> RediStack.RedisClient
     
-    /// Shut down.
-    func shutdown() throws
-    
-    /// Runs a transaction on the redis client using a given closure.
-    ///
-    /// - Parameter transaction: An asynchronous transaction to run on
-    ///   the connection.
-    /// - Returns: The resulting value of the transaction.
-    func transaction<T>(_ transaction: @escaping (RedisProvider) async throws -> T) async throws -> T
+    fileprivate static func configuration(_ config: RedisConnectionPool.Configuration, addresses: [Socket]) -> RedisClient {
+        return RedisClient(provider: ConnectionPool(config: config, lazyAddresses: addresses))
+    }
 }
 
 /// A connection pool is a redis provider with a pool per `EventLoop`.
-private final class ConnectionPool: RedisProvider {
+private final class ConnectionPool: RedisProvider, RediStack.RedisClient {
     /// Map of `EventLoop` identifiers to respective connection pools.
     private var poolStorage: [ObjectIdentifier: RedisConnectionPool] = [:]
     private var poolLock = Lock()
+    private var lazyAddresses: [Socket]?
+    private var logger: Logger?
     
     /// The configuration to create pools with.
     private var config: RedisConnectionPool.Configuration
 
-    init(config: RedisConnectionPool.Configuration) {
+    init(config: RedisConnectionPool.Configuration, lazyAddresses: [Socket]? = nil) {
         self.config = config
+        self.lazyAddresses = lazyAddresses
     }
+    
+    // MARK: - RedisProvider
 
-    func getClient() -> RediStack.RedisClient {
-        getPool()
+    func getClient() throws -> RediStack.RedisClient {
+        try getPool()
     }
 
     func transaction<T>(_ transaction: @escaping (RedisProvider) async throws -> T) async throws -> T {
-        let pool = getPool()
+        let pool = try getPool()
         return try await pool.leaseConnection { conn in
             pool.eventLoop.asyncSubmit { try await transaction(conn) }
         }.get()
@@ -140,17 +187,88 @@ private final class ConnectionPool: RedisProvider {
     ///
     /// - Returns: A `RedisConnectionPool` associated with the current
     ///   `EventLoop` for sending commands to.
-    private func getPool() -> RedisConnectionPool {
+    private func getPool() throws -> RedisConnectionPool {
         let loop = Loop.current
         let key = ObjectIdentifier(loop)
-        return poolLock.withLock {
+        return try poolLock.withLock {
             if let pool = self.poolStorage[key] {
                 return pool
             } else {
-                let newPool = RedisConnectionPool(configuration: self.config, boundEventLoop: loop)
+                var config = self.config
+                if let lazyAddresses = lazyAddresses {
+                    let initialAddresses: [SocketAddress] = try lazyAddresses.map {
+                        switch $0 {
+                        case let .ip(host, port):
+                            return try .makeAddressResolvingHost(host, port: port)
+                        case let .unix(path):
+                            return try .init(unixDomainSocketPath: path)
+                        }
+                    }
+                    
+                    config = RedisConnectionPool.Configuration(
+                        initialServerConnectionAddresses: initialAddresses,
+                        maximumConnectionCount: config.maximumConnectionCount,
+                        connectionFactoryConfiguration: config.factoryConfiguration,
+                        minimumConnectionCount: config.minimumConnectionCount,
+                        connectionBackoffFactor: config.connectionRetryConfiguration.backoff.factor,
+                        initialConnectionBackoffDelay: config.connectionRetryConfiguration.backoff.initialDelay,
+                        connectionRetryTimeout: config.connectionRetryConfiguration.timeout,
+                        poolDefaultLogger: config.poolDefaultLogger)
+                }
+                
+                let newPool = RedisConnectionPool(configuration: config, boundEventLoop: loop)
                 self.poolStorage[key] = newPool
-                return newPool
+                if let logger = logger {
+                    return newPool.logging(to: logger) as? RedisConnectionPool ?? newPool
+                } else {
+                    return newPool
+                }
             }
         }
     }
+    
+    // MARK: RediStack.RedisClient
+    
+    var eventLoop: EventLoop { Loop.current }
+    
+    func logging(to logger: Logger) -> RediStack.RedisClient {
+        self.logger = logger
+        return self
+    }
+    
+    func punsubscribe(from patterns: [String]) -> EventLoopFuture<Void> {
+        wrapError { try getClient().punsubscribe(from: patterns) }
+    }
+    
+    func unsubscribe(from channels: [RedisChannelName]) -> EventLoopFuture<Void> {
+        wrapError { try getClient().unsubscribe(from: channels) }
+    }
+    
+    func send(command: String, with arguments: [RESPValue]) -> EventLoopFuture<RESPValue> {
+        wrapError { try getClient().send(command: command, with: arguments) }
+    }
+    
+    private func wrapError<T>(_ closure: () throws -> EventLoopFuture<T>) -> EventLoopFuture<T> {
+        do { return try closure() }
+        catch { return Loop.current.makeFailedFuture(error) }
+    }
+}
+
+extension RedisConnection: RedisProvider {
+    public func getClient() -> RediStack.RedisClient {
+        self
+    }
+    
+    public func shutdown() throws {
+        try close().wait()
+    }
+    
+    public func transaction<T>(_ transaction: @escaping (RedisProvider) async throws -> T) async throws -> T {
+        try await transaction(self)
+    }
+}
+
+private func wrapError<T>(_ closure: () throws -> EventLoopFuture<T>) -> EventLoopFuture<T> {
+    do { return try closure() }
+    catch { return Loop.current.makeFailedFuture(error) }
 }
diff --git a/Sources/Alchemy/Redis/RedisProvider.swift b/Sources/Alchemy/Redis/RedisProvider.swift
@@ -0,0 +1,21 @@
+import RediStack
+
+/// Under the hood provider for `Redis`. Used so either connection pools
+/// or connections can be injected into `Redis` for accessing redis.
+public protocol RedisProvider {
+    /// Get a redis client for running commands.
+    func getClient() throws -> RediStack.RedisClient
+    
+    /// Log with the given logger.
+    func logging(to logger: Logger) -> RediStack.RedisClient
+    
+    /// Shut down.
+    func shutdown() throws
+    
+    /// Runs a transaction on the redis client using a given closure.
+    ///
+    /// - Parameter transaction: An asynchronous transaction to run on
+    ///   the connection.
+    /// - Returns: The resulting value of the transaction.
+    func transaction<T>(_ transaction: @escaping (RedisProvider) async throws -> T) async throws -> T
+}
diff --git a/Sources/AlchemyTest/Stubs/Redis/StubRedis.swift b/Sources/AlchemyTest/Stubs/Redis/StubRedis.swift
