Repository navigation
Throttle drops input without demand and emits the wrong latest value #29
Description
Throttle on main drops input when downstream demand is zero. It also consumes demand before scheduled output runs, which can prevent latest: true from selecting the latest input.
These behaviors differ from Apple Combine. The comparison below uses OpenCombine at 710ffc9, Apple Combine on macOS 26.6.2, and Swift 6.3.3. Each case uses a one-second throttle interval and the repository's VirtualTimeScheduler. The subscriber returns .none from receive(_:).
Observed differences
| Case | Steps | OpenCombine main |
Apple Combine |
|---|---|---|---|
| No initial demand | Send 1, 2, 3; request .max(1); run scheduled actions |
No output | [3] with latest: true; [1] with latest: false |
| Demand exhausted | Request .max(1); send 1; run scheduled actions; send 2, 3; request .max(1); run scheduled actions |
[1] |
[1, 3] with latest: true; [1, 2] with latest: false |
| Output still pending | Request .max(1); send 1, 2, 3 before running any scheduled actions; run scheduled actions |
[1] with latest: true |
[3] with latest: true |
In the first case, all input and the demand request occur at virtual time zero. Apple Combine emits at t = 0 with latest: true and at t = 1s with latest: false.
Reproduction
Add these methods to ThrottleTests. They use the existing test helpers. Both pass with OPENCOMBINE_COMPATIBILITY_TEST=1 and fail against OpenCombine main.
func testInputBeforeDemandMatchesCombine() throws {
for latest in [false, true] {
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)
XCTAssertEqual(helper.publisher.send(3), .none)
XCTAssertTrue(helper.tracking.inputs.isEmpty)
try XCTUnwrap(helper.downstreamSubscription).request(.max(1))
scheduler.executeScheduledActions()
XCTAssertEqual(Array(helper.tracking.inputs), [latest ? 3 : 1])
}
}
func testLatestInputWithOneDemandMatchesCombine() {
let scheduler = VirtualTimeScheduler()
let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
initialDemand: .max(1),
receiveValueDemand: .none) {
$0.throttle(for: .seconds(1), scheduler: scheduler, latest: true)
}
XCTAssertEqual(helper.publisher.send(1), .none)
XCTAssertEqual(helper.publisher.send(2), .none)
XCTAssertEqual(helper.publisher.send(3), .none)
scheduler.executeScheduledActions()
XCTAssertEqual(Array(helper.tracking.inputs), [3])
}Expected behavior
Input selection and throttle timing should continue to follow Apple Combine when demand is zero or finite. A later demand request should make the selected input available for delivery. Input that arrives before a scheduled output runs should still participate in latest selection.
Context
Extracted from @piercifani's #25
Activity