diff --git a/Sources/OpenCombine/Publishers/Publishers.Throttle.swift b/Sources/OpenCombine/Publishers/Publishers.Throttle.swift index e0f8ec245..e4fe7a972 100644 --- a/Sources/OpenCombine/Publishers/Publishers.Throttle.swift +++ b/Sources/OpenCombine/Publishers/Publishers.Throttle.swift @@ -147,6 +147,7 @@ extension Publishers.Throttle { private var lastEmissionTime: Context.SchedulerTimeType? private var pendingInput: Input? + private var heldInput: Input? private var pendingCompletion: Subscribers.Completion? private var demand: Subscribers.Demand = .none @@ -200,6 +201,9 @@ extension Publishers.Throttle { self.lastTime = lastTime guard demand > .none else { + if latest || heldInput == nil { + heldInput = input + } lock.unlock() return .none } @@ -210,31 +214,36 @@ extension Publishers.Throttle { pendingInput = input lock.unlock() } else if !hasScheduledOutput { - let minimumEmissionTime = - lastEmissionTime.map { $0.advanced(by: interval) } + scheduleEmissionAndUnlock(input, now: lastTime) + } else { + lock.unlock() + } - let emissionTime = - minimumEmissionTime.map { Swift.max(lastTime, $0) } ?? lastTime + return .none + } - demand -= 1 + private func scheduleEmissionAndUnlock(_ input: Input, + now: Context.SchedulerTimeType) { + let minimumEmissionTime = + lastEmissionTime.map { $0.advanced(by: interval) } - pendingInput = input - lock.unlock() + let emissionTime = + minimumEmissionTime.map { Swift.max(now, $0) } ?? now - let action: () -> Void = { [weak self] in - self?.scheduledEmission() - } + demand -= 1 - if emissionTime == lastTime { - scheduler.schedule(action) - } else { - scheduler.schedule(after: emissionTime, action) - } - } else { - lock.unlock() + pendingInput = input + lock.unlock() + + let action: () -> Void = { [weak self] in + self?.scheduledEmission() } - return .none + if emissionTime == now { + scheduler.schedule(action) + } else { + scheduler.schedule(after: emissionTime, action) + } } func receive(completion: Subscribers.Completion) { @@ -319,7 +328,12 @@ extension Publishers.Throttle { return } self.demand += demand - lock.unlock() + guard pendingInput == nil, pendingCompletion == nil, + let input = heldInput.take() else { + lock.unlock() + return + } + scheduleEmissionAndUnlock(input, now: scheduler.now) } func cancel() { diff --git a/Tests/OpenCombineTests/PublisherTests/ThrottleTests.swift b/Tests/OpenCombineTests/PublisherTests/ThrottleTests.swift index bbbccb817..959ed4d57 100644 --- a/Tests/OpenCombineTests/PublisherTests/ThrottleTests.swift +++ b/Tests/OpenCombineTests/PublisherTests/ThrottleTests.swift @@ -519,6 +519,34 @@ final class ThrottleTests: XCTestCase { XCTAssertEqual(subscription.history, [.requested(.unlimited)]) } + func testInputWithoutDemandIsHeldUntilDemand() throws { + for latest in [true, false] { + let scheduler = VirtualTimeScheduler() + let helper = OperatorTestHelper(publisherType: CustomPublisher.self, + initialDemand: nil, + receiveValueDemand: .none) { + $0.throttle(for: .seconds(1), scheduler: scheduler, latest: latest) + } + + XCTAssertEqual(helper.publisher.send(1), .none) + XCTAssertEqual(helper.publisher.send(2), .none) + scheduler.executeScheduledActions() + XCTAssertEqual(helper.tracking.history, [.subscription("Throttle")]) + + try XCTUnwrap(helper.downstreamSubscription).request(.max(1)) + scheduler.executeScheduledActions() + XCTAssertEqual(helper.tracking.history, + [.subscription("Throttle"), .value(latest ? 2 : 1)]) + + XCTAssertEqual(helper.publisher.send(3), .none) + scheduler.executeScheduledActions() + try XCTUnwrap(helper.downstreamSubscription).request(.max(1)) + scheduler.executeScheduledActions() + XCTAssertEqual(helper.tracking.history, + [.subscription("Throttle"), .value(latest ? 2 : 1), .value(3)]) + } + } + func testCancelWhileReceivingInput() throws { let scheduler = VirtualTimeScheduler() let subscription = CustomSubscription()