Detailed Explanation of Producing Your Own Async Sequences with AsyncStream

Photo by Robert Zunikoff on Unsplash

Swift’s AsyncStream is a powerful tool that allows developers to bridge asynchronous, value-producing work (like receiving location updates or incoming WebSocket messages) into the world of Swift Concurrency. It enables seamless integration with async/await patterns and conforms to the AsyncSequence protocol, making it iterable using for await.

Here’s a breakdown of the key concepts and mechanisms behind AsyncStream:

Below is a more in-depth explanation of the concepts and examples you provided about using AsyncStream (and AsyncThrowingStream) in Swift Concurrency. This discussion covers three main ways to create async streams, buffering policies, and how iteration and cancellation work.

1. What is an AsyncStream?

An AsyncStream is a Swift type that lets you create a stream of asynchronous values. You can think of it as a sequence that can produce values over time, with each new value being “awaited” in an async for loop. Unlike a standard sequence, an async stream can suspend while waiting for the next value (for example, waiting on some network or I/O operation), then resume once a new value is ready.

2. Three ways to create an AsyncStream

2.1. Unfolding closure approach

func makeStream(values: Int) -> AsyncStream<String> {
    var valueCount = 0

    return AsyncStream(unfolding: {
        let value = await produceValue(shouldTerminate: valueCount == values)
        valueCount += 1
        return value
    })
}

func produceValue(shouldTerminate: Bool) async -> String? {
    guard !shouldTerminate else {
        return nil
    }
    return UUID().uuidString
}

How it works

• Here, you pass an unfolding closure when you create an AsyncStream.
• Each time the stream wants a new element (for instance, the async for loop awaits the next value), the unfolding closure is called.
• Inside the closure, you can perform asynchronous work (e.g., calling produceValue, which itself might do something time-consuming or external).
• The value returned by the closure is then delivered to whatever is iterating over the stream.
• When the closure eventually returns nil, the stream ends.

Async behavior

• Because the unfolding closure is async, you can await network calls, database requests, or other async logic. This is handy if each next value depends on some asynchronous process.

End of stream

• Once you return nil, no further values are requested or produced. The async for loop that’s consuming the stream will exit.

Cancellation

• If you want to do cleanup when the task that owns the AsyncStream is canceled, you can pass an onCancel closure to AsyncStream. This is only triggered when the task is actually canceled, not when you simply break out of a for await loop.

Throwing streams

• If the operation you’re performing might throw an error, you can use AsyncThrowingStream instead. It’s almost identical, but the iteration must be done with:

for try await value in throwingStream { ... }

• and your unfolding closure can throw.

When to use the unfolding closure

• Good fit: You have a purely async-driven workflow inside the closure, and you only need to yield a new value when the consumer of your stream asks for it.
• Potential drawback: You can’t push new values spontaneously from “the outside.” For example, bridging a delegate-based API (like location updates) can be awkward with this approach because your data might arrive at arbitrary times (not just on-demand).

Continuation-based approach (single closure)

let stream = AsyncStream { continuation in
    print("will start yielding")
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
    print("finished the stream")
}

for await value in stream {
    print("received \(value)")
}

How it works

• In this version, you get a continuation object inside the closure. You can call methods like yield(_:), finish(), or finish(throwing:) (for a throwing stream).
• You can decide exactly when to send (yield) values and when to stop (finish).

Immediate execution

• Notice that the closure (where you yield values) runs as soon as you define stream, not when you begin iterating. Because by default, AsyncStream buffers the yielded values, the code yields 1, 2, and 3 immediately and finishes the stream, all before you even start your for await loop.
• When the for await loop starts, it will consume the buffered values (1, 2, and 3).

Buffering

• By default, all yielded values are buffered, meaning the consumer can still read them later.
• That is why the for-loop sees 1, 2, and 3, even though they were produced before the loop began.

When to use

• This approach is especially handy when you want to push values from your code at arbitrary times, or in response to delegate calls (like location updates, network callbacks, or other event-driven APIs).

2.3. makeStream(of:bufferingPolicy:) approach

A third approach (which you referenced briefly) is essentially a convenience that returns both the stream and its continuation as a tuple. You specify:

let (stream, continuation) = AsyncStream.makeStream(of: Int.self, 
                                                    bufferingPolicy: .bufferingNewest(1))

• This approach is similar to the single closure approach but packaged so you can store the continuation more easily, and produce values from anywhere in your code without capturing the continuation in a closure.
• This makes bridging delegate-based APIs even easier because you can keep that continuation around (e.g., as a property on your delegate or manager).

3. Buffering Policies

When you create an AsyncStream, you can specify how many items it should keep in its buffer. This is crucial when you might yield many values before a consumer starts iterating. Here are the main options:

Unlimited (default)

• If you don’t specify a buffering policy, the stream buffers all yielded values until they are consumed.
• This means the consumer eventually sees every value.

**bufferingNewest(_:)**

• This keeps only the most recent n values in the buffer.
• For example, bufferingNewest(1) means that if you yield several values before the consumer is ready to process them, only the most recent one is kept and the rest are discarded.
• Great for scenarios like location updates, where you only care about the user’s latest location (or the last few).

**bufferingOldest(_:)**

• This policy keeps the first n unconsumed values and discards any new ones until there’s space.
• If you only want to keep the earliest values that were yielded first, but not the new ones once the buffer is full, you’d use this.

**bufferingNewest(0)**

• Effectively means “no buffering.” If a consumer isn’t actively awaiting a value at the time it’s yielded, that value is discarded.
• This can be useful for quick events that are only relevant in the moment (like UI state changes or ephemeral interactions).

4. Iterating over an AsyncStream

Typically, you’ll consume an async stream like so:

for await value in someAsyncStream {
    // Process the value
}

• This for await loop asks the stream for the next value, suspends until the stream provides it, then resumes and executes the loop body.
• If the stream ends (by returning nil or calling finish()), the loop exits.
• If the stream can throw (an AsyncThrowingStream), you’d do:

for try await value in someAsyncThrowingStream {
    // Process the value, or handle errors
}

5. Cancellation

• If the task that’s consuming the stream is canceled, Swift Concurrency will invoke the onCancel closure you provided when creating the stream.
• Important: Exiting a for await loop without canceling the surrounding task does not count as cancellation. So if you break out of a loop normally, you won’t trigger onCancel. However, the next time the loop would fetch a value, it simply won’t happen because you’ve stopped iteration.

6. Typical Usage Patterns

Unfolding closure

• When you want a pull-based stream: the consumer requests values on demand, and you do asynchronous work in the closure each time.

Continuation-based

• When you have an event-based or delegate-based system (e.g., location updates, notifications, or callbacks) that spontaneously produce values. You can store the continuation somewhere and call yield whenever new data arrives.

Buffering policy

• When you only need the most recent value(s) or none at all if missed, you configure the buffer size or strategy accordingly.

7. Summary

• **AsyncStream and `AsyncThrowingStream`** are powerful tools in Swift Concurrency that let you create asynchronous sequences of values.

Unfolding closure:

• Repeatedly calls a closure each time the consumer requests a new value.
• Suited for “pull” scenarios where the consumer is in control.

Continuation-based:

• Gives you a continuation you can call yield on at any time.
• Perfect for “push” scenarios like delegates or event-driven data.

Buffering policies let you control how many values are stored for late subscribers or slow consumers:

• bufferingNewest(n): keep the last n values.
• bufferingOldest(n): keep the first n unconsumed values.
• Default: buffer everything (no limit).

Cancellation only triggers the optional onCancel closure when the task truly cancels, not when you merely stop iterating.

By choosing the right creation style and buffering policy, you can tailor AsyncStream to match all sorts of asynchronous data delivery patterns—whether they are demand-driven (pull) or event-driven (push).