What does an AsyncStream do when the producer yields faster than the consumer reads, and how do the buffering policies differ?

10.12.1 · L5 · Interview

Predict first

Five values are yielded to each policy before anyone reads. What does each consumer see?

let newest = AsyncStream<Int>(bufferingPolicy: .bufferingNewest(2)) { c in
    for i in 1...5 { c.yield(i) }
    c.finish()
}
var got: [Int] = []
for await v in newest { got.append(v) }
print("newest(2):", got)

let oldest = AsyncStream<Int>(bufferingPolicy: .bufferingOldest(2)) { c in
    for i in 1...5 { c.yield(i) }
    c.finish()
}
got = []
for await v in oldest { got.append(v) }
print("oldest(2):", got)

let unbounded = AsyncStream<Int> { c in for i in 1...5 { c.yield(i) }; c.finish() }
got = []
for await v in unbounded.filter({ $0 % 2 == 1 }).map({ $0 * 10 }) { got.append(v) }
print("filtered and mapped:", got)
  • newest(2): [4, 5] / oldest(2): [1, 2] / filtered and mapped: [10, 30, 50]
  • newest(2): [1, 2] / oldest(2): [4, 5] / filtered and mapped: [10, 30, 50]
  • newest(2): [1, 2, 3, 4, 5] / oldest(2): [1, 2, 3, 4, 5] / filtered and mapped: [10, 30, 50]

Mental model. An inbox with a tray size. The unbounded tray never stops growing; “newest” throws away the oldest letter to make room; “oldest” refuses new letters when it is full.

An AsyncStream is a queue between a producer (the Continuation you yield into) and a consumer (for await). When the consumer is slower, the stream has to decide what to keep, and the policy is the bufferingPolicy: .unbounded (the default) keeps everything, which is simple and correct but can grow without limit; .bufferingNewest(n) keeps the last n and discards older ones (right for state updates like “current location” or “latest search results”: only the newest matters); .bufferingOldest(n) keeps the first n and drops newer ones (right when earlier events matter and later ones may be refused). In the check five values are yielded before anyone reads: newest(2) delivers [4, 5], oldest(2) delivers [1, 2].

Know the lifecycle too. yield returns a result telling you whether it was enqueued, dropped or the stream had terminated, so a producer can react. finish() ends the sequence once the buffer has drained. Cancelling the consuming task ends the loop; hook continuation.onTermination to stop the underlying source (remove an observer, stop location updates) when the consumer goes away, and test this path, because a leaked source keeps running. Streams are single-consumer in practice (two loops over one stream split the elements between them), and standard algorithms like filter and map work on them, as the check shows ([10, 30, 50]). For lots of events at high rate, choose the policy deliberately and measure.

Newest keeps the most recent two, oldest keeps the first two, and the default keeps all five.

Common traps

  • Leaving the default unbounded buffer on a high-rate source. Memory grows if the consumer falls behind.
  • Not stopping the source in onTermination. The observer or sensor keeps running after the consumer is gone.

They might follow up with

  • Can two tasks read the same stream? They compete for elements; use a broadcast mechanism if each needs all of them.
  • When do I prefer AsyncThrowingStream? When the source can fail; the error ends the stream and is thrown to the consumer.

Related questions

  • 10.12.2 What happens when a continuation is resumed twice or never, and how do you wrap a callback API correctly?
  • 10.12.3 How do you consume NotificationCenter, delegates and other repeated callbacks as async sequences, and what…
  • 10.6.2 How do you bridge repeated events, GCD and Combine into async code with AsyncStream?
  • 10.6.1 How do continuations bridge callback APIs, and what is the difference between checked and unsafe?

What happens when a continuation is resumed twice or never, and how… →