AsyncSequence

extension AsyncSequence
extension AsyncSequence where Element: Equatable
  • first() Asynchronous

    The first element of the sequence, if there is one.

    Declaration

    Swift

    public func first() async rethrows -> Element?
  • collect() Asynchronous

    Collect all elements from a sequence.

    Declaration

    Swift

    public func collect() async rethrows -> [Element]

    Return Value

    An array of all elements.

  • Consume the async sequence and pass the element’s to a closure.

    let sequence = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

sequence.sink { print($0) }

// Prints:
// 1
// 2
// 3

    Declaration

    Swift

    @discardableResult
public func sink(_ receiveValue: @escaping (Element) -> Void) -> Task<Void, Error>

    Parameters

    receiveValue

    The closure to execute on receipt of a value.

    Return Value

    A task instance.

  • Consume the async sequence and pass the element’s and it’s completion state to two closures.

    let sequence = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish(throwing: TestError())
}

sequence.sink(
    receiveValue: { print("Value: \($0)") },
    receiveCompletion: { print("Complete: \($0)") }
)

// Prints:
// Value: 1
// Value: 2
// Value: 3
// Complete: failure(TestError())

    Declaration

    Swift

    @discardableResult
public func sink(
    receiveValue: @escaping (Element) -> Void,
    receiveCompletion: @escaping (AsyncSequenceCompletion<Error>) -> Void
) -> Task<Void, Never>

    Parameters

    receiveValue

    The closure to execute on receipt of a value.
    receiveCompletion

    The closure to execute on completion.

    Return Value

    A task instance.

Erasure

Combine latest

  • Combine three async sequences.

    The combined sequence emits a tuple of the most-recent elements from each sequence when any of them emit a value.

    If one sequence never emits a value this sequence will finish.

    let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

let streamC = .init { continuation in
    continuation.yield(10)
    continuation.yield(11)
    continuation.finish()
}

for await value in streamA.combineLatest(streamB, streamC) {
    print(value)
}

// Prints:
// (1, 5, 10)
// (2, 6, 11)
// (3, 7, 11)
// (4, 8, 11)
// (4, 9, 11)

    Declaration

    Swift

    public func combineLatest<Q, R>(
    _ q: Q,
    _ r: R
) -> CombineLatest3AsyncSequence<Self, Q, R> where Q: AsyncSequence, R: AsyncSequence

    Parameters

    q

    Another async sequence to combine with.
    r

    Another async sequence to combine with.

    Return Value

    A async sequence combines elements from all sequences.

  • Combine with an additional async sequence to produce a AsyncCombineLatest2Sequence.

    The combined sequence emits a tuple of the most-recent elements from each sequence when any of them emit a value.

    let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in self.streamA.combineLatest(self.streamB) {
    print(value)
}

// Prints:
// (1, 5)
// (2, 6)
// (3, 7)
// (4, 8)
// (4, 9)

    Declaration

    Swift

    public func combineLatest<Q>(
    _ other: Q
) -> CombineLatestAsyncSequence<Self, Q> where Q: AsyncSequence

    Parameters

    other

    Another async sequence to combine with.

    Return Value

    A async sequence combines elements from this and another async sequence.

Throttle

  • Emits elements only after a specified time interval elapses between emissions.

    Use the debounce operator to control the number of values and time between delivery of values from the base async sequence. This operator is useful to process bursty or high-volume async sequences where you need to reduce the number of elements emitted to a rate you specify.

    let stream = AsyncStream<Int> { continuation in
    continuation.yield(0)
    try? await Task.sleep(nanoseconds: 200_000_000)
    continuation.yield(1)
    try? await Task.sleep(nanoseconds: 200_000_000)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.yield(5)
    continuation.finish()
}

for element in try await self.stream.debounce(for: 0.1) {
    print(element)
}

// Prints:
// 0
// 1
// 5

    Declaration

    Swift

    public func debounce(for dueTime: TimeInterval) -> DebounceAsyncSequence<Self>

    Parameters

    base

    The async sequence in which this sequence receives it’s elements.
    dueTime

    The amount of time the async sequence should wait before emitting an element.

    Return Value

    A DebounceAsyncSequence instance.

Merge

  • An asynchronous sequence that merges three async sequences.

    The sequences are iterated through in parallel.

    let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(2)
    continuation.finish()
}

let streamC = .init { continuation in
    continuation.yield(3)
    continuation.finish()
}

for await value in streamA.merge(with: streamB, streamC) {
    print(value)
}

// Prints:
// 1
// 2
// 3
// 4

    Declaration

    Swift

    public func merge(
    with q: Self,
    _ r: Self
) -> Merge3AsyncSequence<Self>

    Parameters

    q

    An async sequence.
    r

    An async sequence.

    Return Value

    A async sequence merges elements from this and another async sequence.

  • An asynchronous sequence that merges two async sequence.

    The sequences are iterated through in parallel.

    let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in streamA.merge(with: streamB) {
    print(value)
}

// Prints:
// 1
// 5
// 2
// 6
// 3
// 7
// 4
// 8
// 9

    Declaration

    Swift

    public func merge(
    with other: Self
) -> MergeAsyncSequence<Self>

    Parameters

    other

    Another async sequence to merge with.

    Return Value

    A async sequence merges elements from this and another async sequence.

Replace error

  • Replaces any errors in the async sequence with the provided element.

    let sequence = Fail<Int, TestError>(
    error: TestError()
)
.replaceError(with: 0)

for await value in stream {
    print(value)
}

// Prints:
// 0

    Declaration

    Swift

    public func replaceError(with output: Element) -> ReplaceErrorAsyncSequence<Self>

    Parameters

    output

    The element with which to replace errors from the base async sequence.

    Return Value

    A ReplaceErrorAsyncSequence instance.

Shared

  • Creates a shareable async sequence that can be used across multiple tasks.

    Declaration

    Swift

    public func shared() -> SharedAsyncSequence<Self>

Throttle

  • Emits either the most-recent or first element emitted by the base async sequence in the specified time interval.

    ThrottleAsyncSequence selectively emits elements from a base async sequence during an interval you specify. Other elements received within the throttling interval aren’t emitted.

    let stream = AsyncStream<Int> { continuation in
    continuation.yield(0)
    try? await Task.sleep(nanoseconds: 100_000_000)
    continuation.yield(1)
    try? await Task.sleep(nanoseconds: 100_000_000)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.yield(5)
    continuation.finish()
}

for element in try await self.stream.throttle(for: 0.05, latest: true) {
    print(element)
}

// Prints:
// 0
// 1
// 2
// 5

    Declaration

    Swift

    public func throttle(for interval: TimeInterval, latest: Bool) -> ThrottleAsyncSequence<Self>

    Parameters

    interval

    The interval in which to emit the most recent element.
    latest

    A Boolean value indicating whether to emit the most recent element.

    Return Value

    A ThrottleAsyncSequence instance.

Zip

  • Create an asynchronous sequence that applys a zip function to the three async sequences.

    Combines the latest elements from three async sequences and emits a tuple.

    The async sequence waits until both provided async sequences have emitted an element, then emits both elements as a tuple.

    If one sequence never emits a value or raises an error then the zipped sequence will finish.

    let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.finish()
}

let streamC = .init { continuation in
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in streamA.zip(streamB, streamC) {
    print(value)
}

// Prints:
// (1, 5, 8)
// (2, 6, 9)

    Declaration

    Swift

    public func zip<Q, R>(
    _ q: Q,
    _ r: R
) -> Zip3AsyncSequence<Self, Q, R> where Q: AsyncSequence, R: AsyncSequence

    Parameters

    q

    Another async sequence.
    r

    Another async sequence.

    Return Value

    A async sequence zips elements from this and another async sequence.

  • Create an asynchronous sequence that applys a zip function to the two async sequences.

    Combines the latest elements from two async sequences and emits a tuple.

    The async sequence waits until both provided async sequences have emitted an element, then emits both elements as a tuple.

    If one sequence never emits a value or raises an error then the zipped sequence will finish.

    let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.finish()
}

for await value in streamA.zip(streamB) {
    print(value)
}

// Prints:
// (1, 5)
// (2, 6)

    Declaration

    Swift

    public func zip<Q>(
    _ other: Q
) -> ZipAsyncSequence<Self, Q> where Q: AsyncSequence

    Parameters

    other

    Another async sequence to zip with.

    Return Value

    A async sequence zips elements from this and another async sequence.

Available where Element: Equatable

  • Emits only elements that don’t match the previous element.

    Declaration

    Swift

    public func removeDuplicates() -> RemoveDuplicatesAsyncSequence<Self>

    Return Value

    A AsyncRemoveDuplicatesSequence instance.

  • Omits any element that the predicate determines is equal to the previous element.

    Declaration

    Swift

    public func removeDuplicates(
    by predicate: @escaping RemoveDuplicatesAsyncSequence<Self>.Predicate
) -> RemoveDuplicatesAsyncSequence<Self>

    Parameters

    predicate

    A closure to evaluate whether two elements are equivalent. Return true from this closure to indicate that the second element is a duplicate of the first.

    Return Value

    A AsyncRemoveDuplicatesSequence instance.