mirror of
https://github.com/swift-server/RediStack.git
synced 2026-05-03 07:32:28 +00:00
Nathan Harris
e7b597fc65
Motivation: One of the great features of Redis is being able to subscribe and receive messages published to specific channels as a way of acting as a message queue for processing jobs. PubSub requires a specific understanding of the connection model that can only be implemented directly in this library. Modifications: - Add: `RedisPubSubHandler` to sit in front of `RedisCommandHandler` to manage subscription callbacks and Redis registration - Add: `publish` and the `pubsub` commands - Add: `addPubSubHandler` extension to `NIO.Channel` - Add: Type-safe String wrapper of `RedisChannelName` for PubSub methods - Add: `pubsubSubscriptionNotFound` error case - Add: `isSubscribed` property to `RedisConnection` - Add: `availableConnectionCount` and `leasedConnectionCount` properties to `RedisConnectionPool` - Add: Metrics for PubSub - Add: `makeNewPool` factory method to `RedisConnectionPoolIntegrationTestCase` - Change: `RedisClient` to require methods for PubSub management, as they are intrinsicly tied to the client's connection model - Change: Parsing of `PING` response for handling special case in PubSub mode - Rename: `ActiveConnectionGauge` to `RedisMetrics.IncrementalGauge` Result: Developers will now be able to use Redis in PubSub mode with both connections and pools. This resolves #6
91 lines
3.2 KiB
Swift
91 lines
3.2 KiB
Swift
//===----------------------------------------------------------------------===//
|
|
//
|
|
// This source file is part of the RediStack open source project
|
|
//
|
|
// Copyright (c) 2020 RediStack project authors
|
|
// Licensed under Apache License v2.0
|
|
//
|
|
// See LICENSE.txt for license information
|
|
// See CONTRIBUTORS.txt for the list of RediStack project authors
|
|
//
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
//
|
|
//===----------------------------------------------------------------------===//
|
|
|
|
import NIO
|
|
import RediStack
|
|
import XCTest
|
|
|
|
/// A helper `XCTestCase` subclass that does the standard work of creating a connection pool to use in test cases.
|
|
///
|
|
/// This is essentially the pooled version of `RedisIntegrationTestCase`
|
|
open class RedisConnectionPoolIntegrationTestCase: XCTestCase {
|
|
/// An overridable value of the Redis instance's hostname to connect to for the test suite(s).
|
|
///
|
|
/// The default value is `RedisConnection.defaultHostname`
|
|
///
|
|
/// This is especially useful to override if you build on Linux & macOS where Redis might be installed locally vs. through Docker.
|
|
open var redisHostname: String { return RedisConnection.defaultHostname }
|
|
|
|
/// The port to connect over to Redis, defaulting to `RedisConnection.defaultPort`.
|
|
open var redisPort: Int { return RedisConnection.defaultPort }
|
|
|
|
/// The password to use to connect to Redis. Default is `nil` - no password authentication.
|
|
open var redisPassword: String? { return nil }
|
|
|
|
public var pool: RedisConnectionPool!
|
|
|
|
public let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 3)
|
|
|
|
deinit {
|
|
do {
|
|
try self.eventLoopGroup.syncShutdownGracefully()
|
|
} catch {
|
|
print("Failed to gracefully shutdown ELG: \(error)")
|
|
}
|
|
}
|
|
|
|
/// Creates a `RediStack.RedisConnectionPool` for the next test case, calling `fatalError` if it was not successful.
|
|
///
|
|
/// See `XCTest.XCTestCase.setUp()`
|
|
open override func setUp() {
|
|
do {
|
|
self.pool = try self.makeNewPool()
|
|
} catch {
|
|
fatalError("Failed to make a RedisConnectionPool: \(error)")
|
|
}
|
|
}
|
|
|
|
/// Sends a "FLUSHALL" command to Redis to clear it of any data from the previous test, then closes the connection.
|
|
///
|
|
/// If any steps fail, a `fatalError` is thrown.
|
|
///
|
|
/// See `XCTest.XCTestCase.tearDown()`
|
|
open override func tearDown() {
|
|
do {
|
|
_ = try self.pool.send(command: "FLUSHALL").wait()
|
|
} catch let err as RedisConnectionPoolError where err == .poolClosed {
|
|
// Ok, this is fine.
|
|
} catch {
|
|
fatalError("Failed to clean up the pool: \(error)")
|
|
}
|
|
|
|
self.pool.close()
|
|
self.pool = nil
|
|
}
|
|
|
|
public func makeNewPool() throws -> RedisConnectionPool {
|
|
let address = try SocketAddress.makeAddressResolvingHost(self.redisHostname, port: self.redisPort)
|
|
let pool = RedisConnectionPool(
|
|
serverConnectionAddresses: [address],
|
|
loop: self.eventLoopGroup.next(),
|
|
maximumConnectionCount: .maximumActiveConnections(4),
|
|
minimumConnectionCount: 0,
|
|
connectionPassword: self.redisPassword
|
|
)
|
|
pool.activate()
|
|
|
|
return pool
|
|
}
|
|
}
|