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
Throttleonmaindrops input when downstream demand is zero. It also consumes demand before scheduled output runs, which can preventlatest: truefrom 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'sVirtualTimeScheduler. The subscriber returns.nonefromreceive(_:).Observed differences
main1, 2, 3; request.max(1); run scheduled actions[3]withlatest: true;[1]withlatest: false.max(1); send1; run scheduled actions; send2, 3; request.max(1); run scheduled actions[1][1, 3]withlatest: true;[1, 2]withlatest: false.max(1); send1, 2, 3before running any scheduled actions; run scheduled actions[1]withlatest: true[3]withlatest: trueIn the first case, all input and the demand request occur at virtual time zero. Apple Combine emits at
t = 0withlatest: trueand att = 1swithlatest: false.Reproduction
Add these methods to
ThrottleTests. They use the existing test helpers. Both pass withOPENCOMBINE_COMPATIBILITY_TEST=1and fail against OpenCombinemain.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
latestselection.Context
Extracted from @piercifani's #25