Structures

An async sequence that performs type erasure by wrapping another async sequence.

If the async sequence that you wish to type erase can throw, then use AnyThrowingAsyncSequenceable.

A throwing async sequence that performs type erasure by wrapping another throwing async sequence.

If the async sequence that you wish to type erase doesn’t throw, then use AnyAsyncSequenceable.

Catches any errors in the async sequence and replaces it with the provided async sequence.

let sequence = Fail<Int, TestError>(
    error: TestError()
)
.catch { error in
    Just(-1)
}

for await value in stream {
    print(value)
}

// Prints:
// -1

An asynchronous sequence that chains two async sequences.

The combined sequence first emits the all the values from the first sequence and then emits all values from the second.

let sequenceA = AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

let sequenceB = AsyncStream<Int> { continuation in
    continuation.yield(4)
    continuation.yield(5)
    continuation.yield(6)
    continuation.finish()
}

let sequenceC = AsyncStream<Int> { continuation in
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in sequenceA <> sequenceB <> sequenceC {
    print(value)
}

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

An asynchronous sequence that combines 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)

An asynchronous sequence that combines two 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()
}

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

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

A async sequence that emits elements only after a specified time interval elapses between emissions.

Use DebounceAsyncSequence async sequence to control the number of values and time between delivery of values from the base async sequence. This async sequence 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

Delays emission of all elements by the provided interval.

let stream = AsyncStream<Int> { continuation in
    continuation.yield(0)
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()
}

let start = Date.now
for element in try await self.stream.delay(for: 0.5) {
    print("\(element) - \(Date.now.timeIntervalSince(start))")
}

// Prints:
// 0 - 0.5
// 1 - 1.0
// 2 - 1.5

An asynchronous sequence that only emits the provided value once.

Empty<Int>().sink(
    receiveValue: { print($0) },
    receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("Finished")
        case .failure:
            print("Failed")
        }
    }
)

// Prints:
// Finished

An asynchronous sequence that immediately throws an error when iterated.

Once the error has been thrown, the iterator will return nil to mark the end of the sequence.

let stream = Fail<Int, TestError>(error: TestError())

do {
    for try await value in stream {
        print(value)
    }
} catch {
    print("Error!")
}

// Prints:
// Error!

An asynchronous sequence that only emits the provided value once.

let stream = Just(1)

for await value in stream {
    print(value)
}

// Prints:
// 1

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 self.streamA.merge(with: self.streamB, self.streamC) {
    print(value)
}

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

An asynchronous sequence that merges two async sequences.

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

A async sequence that broadcasts elements.

let sequence = PassthroughAsyncSequence<Int>()
sequence.yield(0)
sequence.yield(1)
sequence.yield(2)
sequence.finish()

for await value in sequence {
    print(value)
}

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

An asynchronous sequence that streams only elements from the base asynchronous sequence that don’t match the previous element.

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

for await value in stream.removeDuplicates() {
    print(value)
}

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

An async sequence that replaces any errors in the sequence with a provided element.

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

for await value in stream {
    print(value)
}

// Prints:
// 0

An async version of Sequence. Generally used to turn any Sequence into it’s async counterpart.

let sequence = [0, 1, 2, 3].async

for await value in sequence {
    print(value)
}

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

An async sequence that can be shared between multiple tasks.

let values = [
    "a",
    "ab",
    "abc",
    "abcd"
]

let stream = AsyncStream { continuation in
    for value in values {
        continuation.yield(value)
    }
    continuation.finish()
}
.shared()

Task {
    let values = try await self.stream.collect()
    // ...
}

Task.detached {
    let values = try await self.stream.collect()
    // ...
}

let values = try await self.stream.collect()
// ...

An async sequence that emits either the most-recent or first element emitted by the base async sequence in a 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

A async sequence that broadcasts elements.

let sequence = ThrowingPassthroughAsyncSequence<Int>()
sequence.yield(0)
sequence.yield(1)
sequence.yield(2)
sequence.finish(throwing: TestError())

do {
    for try await value in sequence {
      print(value)
    }
} catch {
    print("Error!")
}

// Prints:
// 0
// 1
// 2
// Error!

An asynchronous sequence that applys a zip function to the three async sequences.

Use Zip3AsyncSequence to combine the latest elements from three async sequcnes and emit 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)

An asynchronous sequence that applys a zip function to the two async sequences.

Use ZipAsyncSequence to combine the latest elements from two async sequences and emit 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)