Hi everyone,
This post is to gather feedback and ideas about an operator equivalent to withLatestFrom that we can usually find in the reactive world.
The first point would be to decide if this is relevant from an AsyncSequence paradigm and if we want to go down that road in this repo.
For reference I have already implemented a draft version here: https://github.com/twittemb/swift-async-algorithms/blob/withLatestFrom/Sources/AsyncAlgorithms/AsyncWithLatestFromSequence.swift
with the related unit tests: https://github.com/twittemb/swift-async-algorithms/blob/withLatestFrom/Tests/AsyncAlgorithmsTests/TestWithLatestFrom.swift
The goal of such an operator is to combine 2 AsyncSequences that we can call Base and Other. The characteristics of the resulting AsyncSequence would be:
- the output is a tuple (Base.Element, Other.Element)
- it finishes when Base or Other finishes
- it rethrows when Base or Other throws (if Base and Other are non throwing then the sequence is non throwing)
- when a next Base.Element is captured, then the last known Other.Element is gathered and forms the awaited next tuple.
- when a next Base.Element is captured, and there is not yet an Other.Element, then we wait for Other to produce its first element.
To be able to know the last element from Other at any time, It means that Other is being iterated over independently from Base in its dedicated Task.
Let's take a simplified example from the unit tests to better understand the expected outputs:
var base = GatedSequence([1, 2, 3]) // Sequence that outputs an element only when freed by us
let other = AsyncChannel<String>()
let sequence = base.withLatestFrom(other)
let validator = Validator<(Int, String)>()
validator.test(sequence) { iterator in
let pastEnd = await iterator.next()
XCTAssertNil(pastEnd)
finished.fulfill()
}
var value = await validator.validate()
XCTAssertEqual(value, [])
base.advance() // we free the value 1 in Base
await other.send("a") // we send the value "a" in Other
value = await validator.validate() // we get the next value from sequence
XCTAssertEqual(value, [(1, "a")]) // at this point it behaves like `zip`
base.advance() // we free the value 2 in Base
value = await validator.validate() // get the next value from sequence
XCTAssertEqual(value, [(1, "a"), (2, "a")]) // diverges from `zip` because we don't wait a new value in Other, the last known one is "a"
await other.send("b")
await other.send("c") // we send several values in Other
value = validator.current
XCTAssertEqual(value, [(1, "a"), (2, "a")])
base.advance() // we free the value 3 in Base
value = await validator.validate() // get the next value from sequence
XCTAssertEqual(value, [(1, "a"), (2, "a"), (3, "c")]) // the last known value from Other is now "c"
await other.finish()
wait(for: [finished], timeout: 1.0)
value = validator.current
XCTAssertEqual(value, [(1, "a"), (2, "a"), (3, "c")])
This kind of operator works really well when Other is an AsyncChannel because send() waits for the value to be consumed to un-suspend. It makes the Other's current value predictable although it is iterated over in its dedicated Task.
On the other hand, when Other is a more traditional AsyncSequence, as it is iterated in a Task, we have no control on the pace at which the iteration happens. The Other's current value might not YET be the one we are thinking it will be.
Do you think this is something we could need in the library? in 1.0.0?
Can you think of any edge cases that would make such an operator non viable?
Let's iterate and gather feedback together 
.
Thanks.
.
.