diff --git a/Sources/NIORedis/Commands/ListCommands.swift b/Sources/NIORedis/Commands/ListCommands.swift index 90aae53..d8b90bf 100644 --- a/Sources/NIORedis/Commands/ListCommands.swift +++ b/Sources/NIORedis/Commands/ListCommands.swift @@ -117,6 +117,33 @@ extension RedisClient { public func rpoplpush(from source: String, to dest: String) -> EventLoopFuture { return send(command: "RPOPLPUSH", with: [source, dest]) } + + /// Pops the last element from a source list and pushes it to a destination list, blocking until + /// an element is available from the source list. + /// + /// - Important: + /// This will block the connection from completing further commands until an element + /// is available to pop from the source list. + /// + /// It is **highly** recommended to set a reasonable `timeout` + /// or to use the non-blocking `rpoplpush` method where possible. + /// + /// See [https://redis.io/commands/brpoplpush](https://redis.io/commands/brpoplpush) + /// - Parameters: + /// - source: The key of the list to pop from. + /// - dest: The key of the list to push to. + /// - timeout: The time (in seconds) to wait. `0` means indefinitely. + /// - Returns: The element popped from the source list and pushed to the destination, + /// or `nil` if the timeout was reached. + @inlinable + public func brpoplpush( + from source: String, + to dest: String, + timeout: Int = 0 + ) -> EventLoopFuture { + return send(command: "BRPOPLPUSH", with: [source, dest, timeout]) + .map { $0.isNull ? nil: $0 } + } } // MARK: Insert @@ -250,3 +277,116 @@ extension RedisClient { .mapFromRESP() } } + +// MARK: Blocking Pop + +extension RedisClient { + /// Removes the first element of a list, blocking until an element is available. + /// + /// - Important: + /// This will block the connection from completing further commands until an element + /// is available to pop from the list. + /// + /// It is **highly** recommended to set a reasonable `timeout` + /// or to use the non-blocking `lpop` method where possible. + /// + /// See [https://redis.io/commands/blpop](https://redis.io/commands/blpop) + /// - Parameters: + /// - key: The key of the list to pop from. + /// - Returns: The element that was popped from the list, or `nil` if the timout was reached. + @inlinable + public func blpop(from key: String, timeout: Int = 0) -> EventLoopFuture { + return blpop(from: [key], timeout: timeout) + .map { $0?.1 } + } + + /// Removes the first element of a list, blocking until an element is available. + /// + /// - Important: + /// This will block the connection from completing further commands until an element + /// is available to pop from the group of lists. + /// + /// It is **highly** recommended to set a reasonable `timeout` + /// or to use the non-blocking `lpop` method where possible. + /// + /// See [https://redis.io/commands/blpop](https://redis.io/commands/blpop) + /// - Parameters: + /// - keys: The keys of lists in Redis that should be popped from. + /// - timeout: The time (in seconds) to wait. `0` means indefinitely. + /// - Returns: + /// If timeout was reached, `nil`. + /// + /// Otherwise, the key of the list the element was removed from and the popped element. + @inlinable + public func blpop( + from keys: [String], + timeout: Int = 0 + ) -> EventLoopFuture<(String, RESPValue)?> { + return _bpop(command: "BLPOP", keys, timeout) + } + + /// Removes the last element of a list, blocking until an element is available. + /// + /// - Important: + /// This will block the connection from completing further commands until an element + /// is available to pop from the list. + /// + /// It is **highly** recommended to set a reasonable `timeout` + /// or to use the non-blocking `rpop` method where possible. + /// + /// See [https://redis.io/commands/brpop](https://redis.io/commands/brpop) + /// - Parameters: + /// - key: The key of the list to pop from. + /// - Returns: The element that was popped from the list, or `nil` if the timout was reached. + @inlinable + public func brpop(from key: String, timeout: Int = 0) -> EventLoopFuture { + return brpop(from: [key], timeout: timeout) + .map { $0?.1 } + } + + /// Removes the last element of a list, blocking until an element is available. + /// + /// - Important: + /// This will block the connection from completing further commands until an element + /// is available to pop from the group of lists. + /// + /// It is **highly** recommended to set a reasonable `timeout` + /// or to use the non-blocking `rpop` method where possible. + /// + /// See [https://redis.io/commands/brpop](https://redis.io/commands/brpop) + /// - Parameters: + /// - keys: The keys of lists in Redis that should be popped from. + /// - timeout: The time (in seconds) to wait. `0` means indefinitely. + /// - Returns: + /// If timeout was reached, `nil`. + /// + /// Otherwise, the key of the list the element was removed from and the popped element. + @inlinable + public func brpop( + from keys: [String], + timeout: Int = 0 + ) -> EventLoopFuture<(String, RESPValue)?> { + return _bpop(command: "BRPOP", keys, timeout) + } + + @usableFromInline + func _bpop( + command: String, + _ keys: [String], + _ timeout: Int + ) -> EventLoopFuture<(String, RESPValue)?> { + let args = keys as [RESPValueConvertible] + [timeout] + return send(command: command, with: args) + .flatMapThrowing { + guard !$0.isNull else { return nil } + guard let response = [RESPValue]($0) else { + throw NIORedisError.responseConversion(to: [RESPValue].self) + } + assert(response.count == 2, "Unexpected response size returned!") + guard let key = response[0].string else { + throw NIORedisError.assertionFailure(message: "Unexpected structure in response: \(response)") + } + return (key, response[1]) + } + } +} diff --git a/Tests/NIORedisTests/Commands/ListCommandsTests.swift b/Tests/NIORedisTests/Commands/ListCommandsTests.swift index ab355a9..4aa4525 100644 --- a/Tests/NIORedisTests/Commands/ListCommandsTests.swift +++ b/Tests/NIORedisTests/Commands/ListCommandsTests.swift @@ -102,6 +102,22 @@ final class ListCommandsTests: XCTestCase { XCTAssertEqual(try connection.llen(of: "second").wait(), 1) } + func test_brpoplpush() throws { + _ = try connection.lpush([10], into: "first").wait() + + let element = try connection.brpoplpush(from: "first", to: "second").wait() ?? .null + XCTAssertEqual(Int(element), 10) + + let blockingConnection = try Redis.makeConnection().wait() + let expectation = XCTestExpectation(description: "brpoplpush should never return") + _ = blockingConnection.bzpopmin(from: #function) + .always { _ in expectation.fulfill() } + + let result = XCTWaiter.wait(for: [expectation], timeout: 1) + XCTAssertEqual(result, .timedOut) + try blockingConnection.channel.close().wait() + } + func test_linsert() throws { _ = try connection.lpush([10], into: #function).wait() @@ -129,6 +145,27 @@ final class ListCommandsTests: XCTestCase { XCTAssertEqual(Int(element), 30) } + func test_blpop() throws { + let nilPop = try connection.blpop(from: #function, timeout: 1).wait() + XCTAssertNil(nilPop) + + _ = try connection.lpush([10, 20, 30], into: "first").wait() + let pop1 = try connection.blpop(from: "first").wait() ?? .null + XCTAssertEqual(Int(pop1), 30) + + let pop2 = try connection.blpop(from: ["fake", "first"]).wait() + XCTAssertEqual(pop2?.0, "first") + + let blockingConnection = try Redis.makeConnection().wait() + let expectation = XCTestExpectation(description: "blpop should never return") + _ = blockingConnection.bzpopmin(from: #function) + .always { _ in expectation.fulfill() } + + let result = XCTWaiter.wait(for: [expectation], timeout: 1) + XCTAssertEqual(result, .timedOut) + try blockingConnection.channel.close().wait() + } + func test_lpush() throws { _ = try connection.rpush([10, 20, 30], into: #function).wait() @@ -165,6 +202,27 @@ final class ListCommandsTests: XCTestCase { XCTAssertTrue(result.isNull) } + func test_brpop() throws { + let nilPop = try connection.brpop(from: #function, timeout: 1).wait() + XCTAssertNil(nilPop) + + _ = try connection.lpush([10, 20, 30], into: "first").wait() + let pop1 = try connection.brpop(from: "first").wait() ?? .null + XCTAssertEqual(Int(pop1), 10) + + let pop2 = try connection.brpop(from: ["fake", "first"]).wait() + XCTAssertEqual(pop2?.0, "first") + + let blockingConnection = try Redis.makeConnection().wait() + let expectation = XCTestExpectation(description: "brpop should never return") + _ = blockingConnection.bzpopmin(from: #function) + .always { _ in expectation.fulfill() } + + let result = XCTWaiter.wait(for: [expectation], timeout: 1) + XCTAssertEqual(result, .timedOut) + try blockingConnection.channel.close().wait() + } + func test_rpush() throws { _ = try connection.lpush([10, 20, 30], into: #function).wait() @@ -195,11 +253,14 @@ final class ListCommandsTests: XCTestCase { ("test_lrem", test_lrem), ("test_lrange", test_lrange), ("test_rpoplpush", test_rpoplpush), + ("test_brpoplpush", test_brpoplpush), ("test_linsert", test_linsert), ("test_lpop", test_lpop), + ("test_blpop", test_blpop), ("test_lpush", test_lpush), ("test_lpushx", test_lpushx), ("test_rpop", test_rpop), + ("test_brpop", test_brpop), ("test_rpush", test_rpush), ("test_rpushx", test_rpushx), ]