AsyncExtensions
AsyncExtensions provides a collection of operators, async sequences and async streams that mimics Combine behaviour.
The purpose is to be able to chain operators, just as you would do with any reactive programming framework:
AsyncSequences
.Merge(sequence1, sequence2, sequence3)
.prepend(0)
.handleEvents(onElement: { print($0) }, onFinish: { print("Finished") })
.scan("") { accumulator, element in accumulator + "\(element)" }
.collect { print($0) }
Async Sequences
Async Streams
Operators
- Collect
- Scan
- SwitchToLatest
- FlatMapLatest
- HandleEvents
- Assign
- Multicast
- Share
- WithLatestFrom
- EraseToAnyAsyncSequence
More operators and extensions are to come. Pull requests are of course welcome.
Async Sequences
Just
Just is an AsyncSequence that outputs a single value and finishes.
let justSequence = AsyncSequences.Just<Int>(1)
for try await element in justSequence {
// will be called once with element = 1
}
Empty
Empty is an AsyncSequence that immediately finishes without emitting values.
let emptySequence = AsyncSequences.Empty<Int>()
for try await element in emptySequence {
// will never be called
}
Fail
Fail is an AsyncSequence that outputs no elements and throws an error.
let failSequence = AsyncSequences.Fail<Int, Swift.Error>(error: NSError(domain: "", code: 1))
do {
for try await element in failSequence {
// will never be called
}
} catch {
// will catch `NSError(domain: "", code: 1)` here
}
From
From is an AsyncSequence that outputs elements from a traditional Sequence.
let fromSequence = AsyncSequences.From([1, 2, 3, 4, 5])
for await element in fromSequence {
print(element) // will print 1 2 3 4 5
}
A variation offers to set an interval of time between each element.
let fromSequence = AsyncSequences.From([1, 2, 3, 4, 5], interval: .milliSeconds(10))
for await element in fromSequence {
print(element) // will print 1 2 3 4 5 with an interval of 10ms between elements
}
Merge
Merge is an AsyncSequence that merges several async sequences respecting their temporality while being iterated over. When all the async sequences have finished, so too does the merged async sequence. If an async sequence fails, so too does the merged async sequence.
// 0.1ms 1ms 1.5ms 2ms 3ms 4.5ms
// 4 1 5 2 3 6
let asyncSequence1 = AsyncStream(Int.self, bufferingPolicy: .unbounded) { continuation in
Task {
try await Task.sleep(nanoseconds: 1_000_000)
continuation.yield(1)
try await Task.sleep(nanoseconds: 1_000_000)
continuation.yield(2)
try await Task.sleep(nanoseconds: 1_000_000)
continuation.yield(3)
continuation.finish()
}
}
let asyncSequence2 = AsyncStream(Int.self, bufferingPolicy: .unbounded) { continuation in
Task {
try await Task.sleep(nanoseconds: 100_000)
continuation.yield(4)
try await Task.sleep(nanoseconds: 1_400_000)
continuation.yield(5)
try await Task.sleep(nanoseconds: 3_000_000)
continuation.yield(6)
continuation.finish()
}
}
let mergedAsyncSequence = AsyncSequences.Merge(asyncSequence1, asyncSequence2)
for try await element in mergedAsyncSequence {
print(element) // will print -> 4 1 5 2 3 6
}
Zip2
Zip2 is an AsyncSequence that combines the latest elements from two sequences according to their temporality and emits a tuple to the client. If any async sequence ends successfully or fails with an error, so too does the zipped async Sequence.
let asyncSequence1 = AsyncSequences.From([1, 2, 3, 4, 5])
let asyncSequence2 = AsyncSequences.From(["5", "4", "3", "2", "1"])
let zippedAsyncSequence = AsyncSequences.Zip2(asyncSequence1, asyncSequence2)
for try await element in zippedAsyncSequence {
print(element) // will print -> (1, "5") (2, "4") (3, "3") (4, "2") (5, "1")
}
Zip3
Zip3 is an AsyncSequence that combines the latest elements from two sequences according to their temporality and emits a tuple to the client. If any async sequence ends successfully or fails with an error, so too does the zipped async Sequence.
let asyncSequence1 = AsyncSequences.From([1, 2, 3, 4, 5])
let asyncSequence2 = AsyncSequences.From(["5", "4", "3", "2", "1"])
let asyncSequence3 = AsyncSequences.From([true, false, true, false, true])
let zippedAsyncSequence = AsyncSequences.Zip3(asyncSequence1, asyncSequence2, asyncSequence3)
for try await element in zippedAsyncSequence {
print(element) // will print -> (1, "5", true) (2, "4", false) (3, "3", true) (4, "2", false) (5, "1", true)
}
Zip
Zip is an AsyncSequence that combines the latest elements from several sequences according to their temporality and emits an array to the client. If any async sequence ends successfully or fails with an error, so too does the zipped async Sequence.
let asyncSequence1 = AsyncSequences.From([1, 2, 3])
let asyncSequence2 = AsyncSequences.From([1, 2, 3])
let asyncSequence3 = AsyncSequences.From([1, 2, 3])
let asyncSequence4 = AsyncSequences.From([1, 2, 3])
let asyncSequence5 = AsyncSequences.From([1, 2, 3])
let zippedAsyncSequence = AsyncSequences.Zip(asyncSequence1, asyncSequence2, asyncSequence3, asyncSequence4, asyncSequence5)
for try await element in zippedAsyncSequence {
print(element) // will print -> [1, 1, 1, 1, 1] [2, 2, 2, 2, 2] [3, 3, 3, 3, 3]
}
Timer
Timer is an async sequence that repeatedly emits the current date on the given interval, with the given priority.
let timer = AsyncSequences.Timer(priority: .high, every: .seconds(1))
Task {
for try await element in timer {
print(element)
}
}
// will print:
// 2022-03-06 19:31:22 +0000
// 2022-03-06 19:31:23 +0000
// 2022-03-06 19:31:24 +0000
// 2022-03-06 19:31:25 +0000
// 2022-03-06 19:31:26 +0000
// and will stop once timer.cancel() is called or the parent task is cancelled.
Async Streams
Passthrough
Passthrough is an async sequence in which one can send values over time.
let passthrough = AsyncStreams.Passthrough<Int>()
Task {
for try await element in passthrough {
print(element) // will print 1 2
}
}
Task {
for try await element in passthrough {
print(element) // will print 1 2
}
}
.. later in the application flow
passthrough.send(1)
passthrough.send(2)
CurrentValue
CurrentValue is an async sequence in which one can send values over time. The current value is always accessible as an instance variable. The current value is replayed for any new async loop.
let currentValue = AsyncStreams.CurrentValue<Int>(1)
Task {
for try await element in currentValue {
print(element) // will print 1 2
}
}
Task {
for try await element in currentValue {
print(element) // will print 1 2
}
}
.. later in the application flow
currentValue.send(2)
print(currentValue.element) // will print 2
Replay
Replayis an async sequence in which one can send values over time. Values are buffered in a FIFO fashion so they can be iterated over by new loops. When the bufferSize is outreached the oldest value is dropped.
let replay = AsyncStreams.Replay<Int>(bufferSize: 3)
(1...5).forEach { replay.send($0) }
for try await element in replay {
print(element) // will print 3, 4, 5
}
Streamed
Streamed is a property wrapper that streams a property as an AsyncSequence. It is a structured concurrency equivalent to Combine @Published.
class Weather {
@Streamed var temperature: Double
init(temperature: Double) {
self.temperature = temperature
}
}
let weather = Weather(temperature: 20)
Task {
for try await element in weather.$temperature {
print ("Temperature now: \(element)")
}
}
// ... later in the application flow
weather.temperature = 25
// will print:
// Temperature now: 20.0
// Temperature now: 25.0
Operators
Collect
collect(_:) iterates over each element of the AsyncSequence and give it to the async block.
let fromSequence = AsyncSequences.From([1, 2, 3])
fromSequence
.collect { print($0) } // will print 1 2 3
Scan
scan(_:_:) transforms elements from the upstream async sequence by providing the current element to a closure along with the last value returned by the closure. Each intermediate value will be emitted in the downstream async sequence.
let sourceSequence = AsyncSequences.From([1, 2, 3, 4, 5])
let scannedSequence = sourceSequence.scan("") { accumulator, element in accumulator + "\(element)"}
for try await element in scannedSequence {
print(element)
}
// will print:
1
12
123
1234
12345
SwitchToLatest
switchToLatest() re-emits elements sent by the most recently received async sequence. This operator applies only in the case where the upstream async sequence's Element is it-self an async sequence.
let sourceSequence = AsyncSequences.From([1, 2, 3])
let mappedSequence = sourceSequence.map { element in
AsyncSequences.From(["a\(element)", "b\(element)"])
}
let switchedSequence = mappedSequence.switchToLatest()
for try await element in switchedSequence {
print(element) // will print a3 b3
}
FlatMapLatest
flatMapLatest(_:) transforms the upstream async sequence elements into an async sequence and flattens the sequence of events from these multiple sources async sequences to appear as if they were coming from a single async sequence of events. Mapping to a new async sequence will cancel the task related to the previous one.
This operator is basically a shortcut for map() and switchToLatest().
let sourceSequence = AsyncSequences.From([1, 2, 3])
let flatMapLatestSequence = sourceSequence.flatMapLatest { element in
AsyncSequences.From(["a\(element)", "b\(element)"])
}
for try await element in flatMapLatestSequence {
print(element) // will print a3 b3
}
Prepend
prepend(_:) prepends an element to the upstream async sequence.
let sourceSequence = AsyncSequences.From([1, 2, 3])
let prependSequence = sourceSequence.prepend(0)
for try await element in prependSequence {
print(element) // will print 0 1 2 3
}
HandleEvents
handleEvents(onStart:onElement:onCancel:onFinish) performs the specified closures when async sequences events occur.
let sourceSequence = AsyncSequences.From([1, 2, 3, 4, 5])
let handledSequence = sourceSequence.handleEvents {
print("Begin iterating")
} onElement: { element in
print("Element is \(element)")
} onCancel: {
print("Cancelled")
} onFinish: { termination in
print(termination)
}
for try await element in handledSequence {}
// will print:
// Begin iterating
// Element is 1
// Element is 2
// Element is 3
// Element is 4
// Element is 5
// finished
Assign
assign(to:on:) assigns each element from the async sequence to a property on an object.
class Root {
var property: String = ""
}
let root = Root()
let fromSequence = AsyncSequences.From(["1", "2", "3"])
try await fromSequence.assign(to: \.property, on: root) // will set the property value to "1", "2", "3"
Multicast
multicast(_:) is useful when you have multiple client loops, but you want the upstream async sequence to only produce a single AsyncIterator.
let stream = AsyncStreams.Passthrough<(String, Int)>()
let multicastedAsyncSequence = ["First", "Second", "Third"]
.asyncElements
.map { ($0, Int.random(in: 0...100)) }
.handleEvents(onElement: { print("AsyncSequence produces: ($0)") })
.multicast(stream)
Task {
try await multicastedAsyncSequence
.collect { print ("Task 1 received: \($0)") }
}
Task {
try await multicastedAsyncSequence
.collect { print ("Task 2 received: \($0)") }
}
multicastedAsyncSequence.connect()
// will print:
// AsyncSequence produces: ("First", 78)
// Stream 2 received: ("First", 78)
// Stream 1 received: ("First", 78)
// AsyncSequence produces: ("Second", 98)
// Stream 2 received: ("Second", 98)
// Stream 1 received: ("Second", 98)
// AsyncSequence produces: ("Third", 61)
// Stream 2 received: ("Third", 61)
// Stream 1 received: ("Third", 61)
Share
share() shares the output of an upstream async sequence with multiple client loops. share() is effectively a shortcut for multicast(_:) using a Passthrough stream, with an implicit autoconnect().
let sharedAsyncSequence = AsyncSequences.From(["first", "second", "third"], interval: .seconds(1))
.map { ($0, Int.random(in: 0...100)) }
.handleEvents(onElement: { print("AsyncSequence produces: \($0)") })
.share()
Task {
try await sharedAsyncSequence
.collect { print ("Task 1 received: \($0)") }
}
Task {
try await sharedAsyncSequence
.collect { print ("Task 2 received: \($0)") }
}
// will print:
// AsyncSequence produces: ("First", 78)
// Stream 2 received: ("First", 78)
// Stream 1 received: ("First", 78)
// AsyncSequence produces: ("Second", 98)
// Stream 2 received: ("Second", 98)
// Stream 1 received: ("Second", 98)
// AsyncSequence produces: ("Third", 61)
// Stream 2 received: ("Third", 61)
// Stream 1 received: ("Third", 61)
WithLatestFrom
withLatestFrom(_:) merges two async sequences into a single one by combining each value from self with the latest value from the other sequence, if any.
let seq1 = AsyncStreams.CurrentValue<Int>(1)
let seq2 = AsyncStreams.CurrentValue<String>("1")
let combinedSeq = seq1.withLatestFrom(seq2)
Task {
for try await element in combinedSeq {
print(element)
}
}
seq1.send(2)
seq2.send("2")
seq1.send(3)
// will print:
(1, "1")
(2, "1")
(3, "2")
EraseToAnyAsyncSequence
eraseToAnyAsyncSequence() type-erases the async sequence into an AnyAsyncSequence.
![[BUG] Latest Main branch commit breaks AsyncStreams (CurrentValue, Passthrough etc)](https://avatars.githubusercontent.com/u/2805285?v=4)
48 Dec 4, 2022
2k Dec 24, 2022
23 Oct 14, 2022
814 Oct 30, 2022
684 Jan 9, 2023
1.6k Oct 17, 2022
1 Dec 26, 2021
89 Dec 30, 2022
33 Dec 27, 2022
2 Jul 13, 2022
1 Nov 30, 2021
7 Oct 2, 2022
11 Feb 1, 2022
17 Dec 11, 2022
3 Nov 23, 2022
0 Jan 2, 2022
9 Oct 4, 2022