-
Notifications
You must be signed in to change notification settings - Fork 26
/
AsyncMergeSequence.swift
57 lines (48 loc) · 1.5 KB
/
AsyncMergeSequence.swift
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
//
// AsyncMergeSequence.swift
//
//
// Created by Thibault Wittemberg on 31/03/2022.
//
/// Creates an asynchronous sequence of elements from many underlying asynchronous sequences
public func merge<Base: AsyncSequence>(
_ bases: Base...
) -> AsyncMergeSequence<Base> {
AsyncMergeSequence(bases)
}
/// An asynchronous sequence of elements from many underlying asynchronous sequences
///
/// In a `AsyncMergeSequence` instance, the *i*th element is the *i*th element
/// resolved in sequential order out of the two underlying asynchronous sequences.
/// Use the `merge(...)` function to create an `AsyncMergeSequence`.
public struct AsyncMergeSequence<Base: AsyncSequence>: AsyncSequence {
public typealias Element = Base.Element
public typealias AsyncIterator = Iterator
let bases: [Base]
public init(_ bases: [Base]) {
self.bases = bases
}
public func makeAsyncIterator() -> Iterator {
Iterator(
bases: self.bases
)
}
public struct Iterator: AsyncIteratorProtocol {
let mergeStateMachine: MergeStateMachine<Element>
init(bases: [Base]) {
self.mergeStateMachine = MergeStateMachine(
bases
)
}
public mutating func next() async rethrows -> Element? {
let mergedElement = await self.mergeStateMachine.next()
switch mergedElement {
case .element(let result):
return try result._rethrowGet()
case .termination:
return nil
}
}
}
}
extension AsyncMergeSequence: Sendable where Base: Sendable {}