Skip to content

Ensure the last element of reduction in the throttle is emitted and use appropriate delay #292

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Sep 21, 2023
Merged

Ensure the last element of reduction in the throttle is emitted and use appropriate delay #292

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion Sources/AsyncAlgorithms/AsyncThrottleSequence.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,16 @@ extension AsyncThrottleSequence: AsyncSequence {
let start = last ?? clock.now
repeat {
guard let element = try await base.next() else {
return nil
if reduced != nil, let last {
// ensure the rate of elements never exceeds the given interval
let amount = interval - last.duration(to: clock.now)
if amount > .zero {
try? await clock.sleep(for: amount)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we swallowing the cancellation error here?

Copy link
Member Author

@phausler phausler Sep 18, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it cannot throw since that would alter the signature from rethrowing to flat-out throwing.

I guess it could return nil immediately in that case.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wondering if we should throw in the case where we are a throwing sequence and if not just return nil.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can't really know that - plus throwing CancellationError seems rather off to me except in the cases where EOF and cancellation need disambiguation

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is impossible to differentiate when a non-throwing async sequence returns nil if it was due to EOF or task cancellation, that's why I am leaning towards always throwing the CancellationError if possible. Now this case is interesting because we would just return the reduced value and probably the next iteration will potentially throw the CancellationError. So I guess it's fine

}
}
// the last value is unable to have any subsequent
// values so always return the last reduction
return reduced
}
let reduction = await reducing(reduced, element)
let now = clock.now
Expand Down
22 changes: 20 additions & 2 deletions Tests/AsyncAlgorithmsTests/TestThrottle.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ final class TestThrottle: XCTestCase {
validate {
"abcdefghijk|"
$0.inputs[0].throttle(for: .steps(3), clock: $0.clock)
"a--d--g--j-|"
"a--d--g--j--[k|]"
}
}

Expand All @@ -81,7 +81,7 @@ final class TestThrottle: XCTestCase {
validate {
"abcdefghijk|"
$0.inputs[0].throttle(for: .steps(3), clock: $0.clock, latest: false)
"a--b--e--h-|"
"a--b--e--h--[k|]"
}
}

Expand Down Expand Up @@ -138,4 +138,22 @@ final class TestThrottle: XCTestCase {
"-a---c---e---g---i---k-|"
}
}

func test_trailing_delay_without_latest() throws {
guard #available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *) else { throw XCTSkip("Skipped due to Clock/Instant/Duration availability") }
validate {
"abcdefghijkl|"
$0.inputs[0].throttle(for: .steps(3), clock: $0.clock, latest: false)
"a--b--e--h--[k|]"
}
}

func test_trailing_delay_with_latest() throws {
guard #available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *) else { throw XCTSkip("Skipped due to Clock/Instant/Duration availability") }
validate {
"abcdefghijkl|"
$0.inputs[0].throttle(for: .steps(3), clock: $0.clock, latest: true)
"a--d--g--j--[l|]"
}
}
}