Throttle: hold input received without downstream demand

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<Failure>?
 
         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<Failure>) {
@@ -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() {

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()