Add subscribe and psubscribe overloads to RedisConnectionPoolService (#…

…27)
hummingbird-project · Sep 6, 2024 · 316d8b4 · 316d8b4
1 parent 0f83aae
commit 316d8b4
Show file tree

Hide file tree

Showing 2 changed files with 44 additions and 1 deletion.
diff --git a/Sources/HummingbirdRedis/RedisConnectionPoolService.swift b/Sources/HummingbirdRedis/RedisConnectionPoolService.swift
@@ -138,6 +138,26 @@ extension RedisConnectionPoolService: RedisClient {
         self.pool.logging(to: logger)
     }
 
+    @inlinable
+    public func subscribe(
+        to channels: [RedisChannelName],
+        messageReceiver receiver: @escaping RedisSubscriptionMessageReceiver,
+        onSubscribe subscribeHandler: RedisSubscriptionChangeHandler?,
+        onUnsubscribe unsubscribeHandler: RedisSubscriptionChangeHandler?
+    ) -> EventLoopFuture<Void> {
+        self.pool.subscribe(to: channels, messageReceiver: receiver, onSubscribe: subscribeHandler, onUnsubscribe: unsubscribeHandler)
+    }
+
+    @inlinable
+    public func psubscribe(
+        to patterns: [String],
+        messageReceiver receiver: @escaping RedisSubscriptionMessageReceiver,
+        onSubscribe subscribeHandler: RedisSubscriptionChangeHandler?,
+        onUnsubscribe unsubscribeHandler: RedisSubscriptionChangeHandler?
+    ) -> EventLoopFuture<Void> {
+        self.pool.psubscribe(to: patterns, messageReceiver: receiver, onSubscribe: subscribeHandler, onUnsubscribe: unsubscribeHandler)
+    }
+
     @inlinable
     public func unsubscribe(from channels: [RediStack.RedisChannelName]) -> NIOCore.EventLoopFuture<Void> {
         self.pool.unsubscribe(from: channels)

diff --git a/Tests/HummingbirdRedisTests/RedisTests.swift b/Tests/HummingbirdRedisTests/RedisTests.swift
@@ -23,7 +23,7 @@ final class HummingbirdRedisTests: XCTestCase {
     static let env = Environment()
     static let redisHostname = env.get("REDIS_HOSTNAME") ?? "localhost"
 
-    func testApplication() async throws {
+    func testConnectionPoolService() async throws {
         let redis = try RedisConnectionPoolService(
             .init(hostname: Self.redisHostname, port: 6379),
             logger: Logger(label: "Redis")
@@ -35,6 +35,29 @@ final class HummingbirdRedisTests: XCTestCase {
         try await redis.close()
     }
 
+    func testSubscribe() async throws {
+        let expectation = XCTestExpectation(description: "Waiting on subscription")
+        expectation.expectedFulfillmentCount = 1
+        let redis = try RedisConnectionPoolService(
+            .init(hostname: Self.redisHostname, port: 6379),
+            logger: Logger(label: "Redis")
+        )
+        let redis2 = try RedisConnectionPoolService(
+            .init(hostname: Self.redisHostname, port: 6379),
+            logger: Logger(label: "Redis")
+        )
+
+        _ = try await redis.subscribe(to: ["channel"]) { _, value in
+            XCTAssertEqual(value, .init(from: "hello"))
+            expectation.fulfill()
+        }.get()
+        _ = try await redis2.publish("hello", to: "channel").get()
+        await fulfillment(of: [expectation], timeout: 5)
+        _ = try await redis.unsubscribe(from: ["channel"]).get()
+        try await redis.close()
+        try await redis2.close()
+    }
+
     func testRouteHandler() async throws {
         let redis = try RedisConnectionPoolService(
             .init(hostname: Self.redisHostname, port: 6379),