feat: Add AsyncSequence.flatMapLatest operator#382
Conversation
This introduces a simplified version of the flatMapLatest operator for AsyncSequence, along with a basic unit test. Note that this implementation uses unstructured concurrency and may have race conditions regarding task cancellation.
- Replaces naive implementation with a thread-safe approach using ManagedCriticalState. - Introduces generation tracking to prevent race conditions where cancelled inner sequences could yield stale values. - Adds test_interleaving_race_condition to verify correctness under concurrent load. - Ensures Swift 6 Sendable compliance.
- Replaces AsyncThrowingStream implementation with a custom AsyncSequence, Storage, and StateMachine. - Implements explicit state management using Lock for thread safety. - Handles concurrency between outer and inner sequences robustly. - Ensures correct cancellation propagation and error handling. - Verified with test_interleaving_race_condition.
- Moves AsyncFlatMapLatestSequence.swift to Sources/AsyncAlgorithms/FlatMapLatest/ - Extracts FlatMapLatestStateMachine and FlatMapLatestStorage into their own files. - Aligns project structure with other complex operators like CombineLatest and Debounce.
- Added test_outer_throwing to verify outer sequence error propagation - Added test_inner_throwing to verify inner sequence error propagation - Added test_cancellation to verify proper cancellation handling - Added test_empty_outer to verify empty outer sequence handling - Added test_empty_inner to verify empty inner sequence handling - Fixed test_simple_sequence to be more robust against timing issues
- Changed from synchronous map with throwIf to AsyncThrowingStream - Added delay to ensure proper error propagation timing - Fixes intermittent test failures
- Removed LLM-style thinking comments - Kept only essential, professional explanations - No functional changes, all tests pass
- Added FlatMapLatest.md documentation guide - Authored by Peter Friese - Includes introduction, code samples, and use cases - Covers search-as-you-type, location-based data, and dynamic config examples - Compares with similar operators in ReactiveX and Combine
- Created SAA-00nn proposal for flatMapLatest operator - Includes motivation, detailed design, and examples - Covers implementation strategy with state machine - Compares with ReactiveX switchMap and Combine switchToLatest
phausler
left a comment
phausler
left a comment
There was a problem hiding this comment.
There are more detailed review notes that can be made but these are perhaps the most impactful for the pitch phase.
| self.storage = FlatMapLatestStorage(base: base, transform: transform) | ||
| } | ||
|
|
||
| public func next() async throws -> Element? { |
There was a problem hiding this comment.
couple of notes here that might inform the rest of the potential pitch:
the next method should likely be the typed throws variant (because as it currently stands this algorithm always throws, which is less than ideal if the base components never throw)
that would then make the overload for the old variant of next (the unaddorned non-typed throws version as you have written) would be rethrows not throws
There was a problem hiding this comment.
+1 on using typed throws, thanks for the suggestion! I see you implemented this in PR #395, so I will not change it here for the time being.
There was a problem hiding this comment.
so my pr was just to update things, feel free to merge my branch into yours and continue iterating. the major concerns here are more for the review with regards to the closure throwing or being async etc. I can help out with getting the implementation over the line.
There was a problem hiding this comment.
I cherry-picked your suggestions into this PR, thanks!
| } | ||
|
|
||
| enum Action { | ||
| case startInnerTask(Inner, generation: Int, previousTask: Task<Void, Never>?, previousContinuation: UnsafeContinuation<Void, Error>?) |
There was a problem hiding this comment.
I would like to potentially have the flatMap and switchToLatest families eventually only be driven by one task, do you think that is something we could transition to from this implementation?
There was a problem hiding this comment.
I tried following the implementation of the other algorithms.
Maybe this is something we can look into once the initial version has been merged?
There was a problem hiding this comment.
yes this could be an optimization later down the road.
* Implement AsyncMapErrorSequence * Update the proposal, implementation and tests with formatting, additional details and modifications for being able to push it to a review * Foratting and preambles * Update Copyright date for error sequence Co-authored-by: Clive <clive819@users.noreply.github.com> * Update copyright year for mapError tests Co-authored-by: Clive <clive819@users.noreply.github.com> * Update availablity and modify the mapError algorithm to be opaque * Correct pre 6.0 build avail parameters * Make the formatting of the algorithm accesor a bit more consistent * Update the 5.8 package builder to define 1.2 * re-run formatter --------- Co-authored-by: Clive <clive819@gmail.com> Co-authored-by: Clive <clive819@users.noreply.github.com>
peterfriese
force-pushed
the
flatMapLatest/statemachine
branch
from
e775682 to
cf88997
Compare
|
So there are a few formatting issues and housekeeping tasks I can take care of - namely I will be marking this as 1.3 (but same OS requirement set) to keep track of the versioning. All in all this looks great! thanks for the hard work on getting this over the line. |
2 participants
This PR introduces the
flatMapLatestoperator forAsyncSequence, refined through collaboration with Philippe Hausler.What's included:
flatMapLatestoperator: A new operator that transforms elements of an asynchronous sequence into new asynchronous sequences, automatically cancelling previous inner sequences when a new element arrives from the outer sequence.some AsyncSequence) for better encapsulation.switchToLatest()shorthand forAsyncSequenceofAsyncSequence.test_concurrency_crash_repro) to ensure the race condition fix remains effective.Evolution/00nn-flatMapLatest.md) reflecting the pitchable state with the refined API.mapErroroperator (dependency) to support the refined implementation.Motivation:
The
flatMapLatestoperator (and its shorthandswitchToLatest) addresses the common problem of managing asynchronous operations where only the result of the most recent operation is relevant. This is essential for building responsive, data-driven user interfaces in Swift.This PR is an implementation of #381
This PR represents an integrated effort combining the original proposal with refinements for modern Swift Concurrency patterns.