1# Async Sequences and Streams
2 
3Use this when:
4 
5- You need to iterate over values that arrive over time.
6- You are bridging callback-based or delegate-based APIs to async/await.
7- You need to choose between `AsyncSequence`, `AsyncStream`, or a regular async method.
8 
9Skip this file if:
10 
11- You need time-based operators like debounce, throttle, or merge. Use `async-algorithms.md`.
12- You are choosing between `Task`, `async let`, or task groups. Use `tasks.md`.
13 
14Jump to:
15 
16- AsyncSequence Protocol
17- AsyncStream / AsyncThrowingStream
18- Bridging Callbacks and Delegates
19- Stream Lifecycle and Cleanup
20- Buffer Policies
21- Standard Library Integration
22- Limitations
23- When to Use AsyncAlgorithms
24 
25## AsyncSequence
26 
27Protocol for asynchronous iteration over values that become available over time.
28 
29### Basic usage
30 
31```swift
32for await value in someAsyncSequence {
33    print(value)
34}
35```
36 
37**Key difference from Sequence**: Values may not all be available immediately.
38 
39### Custom implementation
40 
41```swift
42struct Counter: AsyncSequence, AsyncIteratorProtocol {
43    typealias Element = Int
44    
45    let limit: Int
46    var current = 1
47    
48    mutating func next() async -> Int? {
49        guard !Task.isCancelled else { return nil }
50        guard current <= limit else { return nil }
51        
52        let result = current
53        current += 1
54        return result
55    }
56    
57    func makeAsyncIterator() -> Counter {
58        self
59    }
60}
61 
62// Usage
63for await count in Counter(limit: 5) {
64    print(count) // 1, 2, 3, 4, 5
65}
66```
67 
68### Standard operators
69 
70Same functional operators as regular sequences:
71 
72```swift
73// Filter
74for await even in Counter(limit: 5).filter({ $0 % 2 == 0 }) {
75    print(even) // 2, 4
76}
77 
78// Map
79let mapped = Counter(limit: 5).map { $0 % 2 == 0 ? "Even" : "Odd" }
80for await label in mapped {
81    print(label)
82}
83 
84// Contains (awaits until found or sequence ends)
85let contains = await Counter(limit: 5).contains(3) // true
86```
87 
88### Termination
89 
90Return `nil` from `next()` to end iteration:
91 
92```swift
93mutating func next() async -> Int? {
94    guard !Task.isCancelled else {
95        return nil // Stop on cancellation
96    }
97    
98    guard current <= limit else {
99        return nil // Stop at limit
100    }
101    
102    return current
103}
104```
105 
106> **Course Deep Dive**: This topic is covered in detail in [Lesson 6.1: Working with asynchronous sequences](https://www.swiftconcurrencycourse.com?utm_source=github&utm_medium=agent-skill&utm_campaign=lesson-reference)
107 
108## AsyncStream
109 
110Convenient way to create async sequences without implementing protocols.
111 
112### Basic creation
113 
114```swift
115let stream = AsyncStream<Int> { continuation in
116    for i in 1...5 {
117        continuation.yield(i)
118    }
119    continuation.finish()
120}
121 
122for await value in stream {
123    print(value)
124}
125```
126 
127### AsyncThrowingStream
128 
129For streams that can fail:
130 
131```swift
132let throwingStream = AsyncThrowingStream<Int, Error> { continuation in
133    continuation.yield(1)
134    continuation.yield(2)
135    continuation.finish(throwing: SomeError())
136}
137 
138do {
139    for try await value in throwingStream {
140        print(value)
141    }
142} catch {
143    print("Error: \(error)")
144}
145```
146 
147> **Course Deep Dive**: This topic is covered in detail in [Lesson 6.2: Using AsyncStream and AsyncThrowingStream in your code](https://www.swiftconcurrencycourse.com?utm_source=github&utm_medium=agent-skill&utm_campaign=lesson-reference)
148 
149## Bridging Closures to Streams
150 
151### Progress + completion handlers
152 
153```swift
154// Old closure-based API
155struct FileDownloader {
156    enum Status {
157        case downloading(Float)
158        case finished(Data)
159    }
160    
161    func download(
162        _ url: URL,
163        progressHandler: @escaping (Float) -> Void,
164        completion: @escaping (Result<Data, Error>) -> Void
165    ) throws {
166        // Implementation
167    }
168}
169 
170// Modern stream-based API
171extension FileDownloader {
172    func download(_ url: URL) -> AsyncThrowingStream<Status, Error> {
173        AsyncThrowingStream { continuation in
174            do {
175                try self.download(url, progressHandler: { progress in
176                    continuation.yield(.downloading(progress))
177                }, completion: { result in
178                    switch result {
179                    case .success(let data):
180                        continuation.yield(.finished(data))
181                        continuation.finish()
182                    case .failure(let error):
183                        continuation.finish(throwing: error)
184                    }
185                })
186            } catch {
187                continuation.finish(throwing: error)
188            }
189        }
190    }
191}
192 
193// Usage
194for try await status in downloader.download(url) {
195    switch status {
196    case .downloading(let progress):
197        print("Progress: \(progress)")
198    case .finished(let data):
199        print("Done: \(data.count) bytes")
200    }
201}
202```
203 
204### Simplified with Result
205 
206```swift
207AsyncThrowingStream { continuation in
208    try self.download(url, progressHandler: { progress in
209        continuation.yield(.downloading(progress))
210    }, completion: { result in
211        continuation.yield(with: result.map { .finished($0) })
212        continuation.finish()
213    })
214}
215```
216 
217## Bridging Delegates
218 
219### Location updates example
220 
221```swift
222final class LocationMonitor: NSObject {
223    private var continuation: AsyncThrowingStream<CLLocation, Error>.Continuation?
224    let stream: AsyncThrowingStream<CLLocation, Error>
225    
226    override init() {
227        var capturedContinuation: AsyncThrowingStream<CLLocation, Error>.Continuation?
228        stream = AsyncThrowingStream { continuation in
229            capturedContinuation = continuation
230        }
231        super.init()
232        self.continuation = capturedContinuation
233        
234        locationManager.delegate = self
235        locationManager.startUpdatingLocation()
236    }
237}
238 
239extension LocationMonitor: CLLocationManagerDelegate {
240    func locationManager(_ manager: CLLocationManager, didUpdateLocations locations: [CLLocation]) {
241        for location in locations {
242            continuation?.yield(location)
243        }
244    }
245    
246    func locationManager(_ manager: CLLocationManager, didFailWithError error: Error) {
247        continuation?.finish(throwing: error)
248    }
249}
250 
251// Usage
252let monitor = LocationMonitor()
253for try await location in monitor.stream {
254    print("Location: \(location.coordinate)")
255}
256```
257 
258## Stream Lifecycle
259 
260### Termination callback
261 
262```swift
263AsyncThrowingStream<Int, Error> { continuation in
264    continuation.onTermination = { @Sendable reason in
265        print("Terminated: \(reason)")
266        // Cleanup: remove observers, cancel work, etc.
267    }
268    
269    continuation.yield(1)
270    continuation.finish()
271}
272```
273 
274**Termination reasons**:
275- `.finished` - Normal completion
276- `.finished(Error?)` - Completed with error (throwing stream)
277- `.cancelled` - Task canceled
278 
279### Cancellation
280 
281Streams cancel when:
282- Enclosing task cancels
283- Stream goes out of scope
284 
285```swift
286let task = Task {
287    for try await status in download(url) {
288        print(status)
289    }
290}
291 
292task.cancel() // Triggers onTermination with .cancelled
293```
294 
295**No explicit cancel method** - rely on task cancellation.
296 
297## Buffer Policies
298 
299Control what happens to values when no one is awaiting:
300 
301### .unbounded (default)
302 
303Buffers all values until consumed:
304 
305```swift
306let stream = AsyncStream<Int> { continuation in
307    (0...5).forEach { continuation.yield($0) }
308    continuation.finish()
309}
310 
311try await Task.sleep(for: .seconds(1))
312 
313for await value in stream {
314    print(value) // Prints all: 0, 1, 2, 3, 4, 5
315}
316```
317 
318### .bufferingNewest(n)
319 
320Keeps only the newest N values:
321 
322```swift
323let stream = AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in
324    (0...5).forEach { continuation.yield($0) }
325    continuation.finish()
326}
327 
328try await Task.sleep(for: .seconds(1))
329 
330for await value in stream {
331    print(value) // Prints only: 5
332}
333```
334 
335### .bufferingOldest(n)
336 
337Keeps only the oldest N values:
338 
339```swift
340let stream = AsyncStream(bufferingPolicy: .bufferingOldest(1)) { continuation in
341    (0...5).forEach { continuation.yield($0) }
342    continuation.finish()
343}
344 
345try await Task.sleep(for: .seconds(1))
346 
347for await value in stream {
348    print(value) // Prints only: 0
349}
350```
351 
352### .bufferingNewest(0)
353 
354Only receives values emitted after iteration starts:
355 
356```swift
357let stream = AsyncStream(bufferingPolicy: .bufferingNewest(0)) { continuation in
358    continuation.yield(1) // Discarded
359    
360    Task {
361        try await Task.sleep(for: .seconds(2))
362        continuation.yield(2) // Received
363        continuation.finish()
364    }
365}
366 
367try await Task.sleep(for: .seconds(1))
368 
369for await value in stream {
370    print(value) // Prints only: 2
371}
372```
373 
374**Use case**: Location updates, file system changes - only care about latest.
375 
376## Repeated Async Calls
377 
378Use `init(unfolding:onCancel:)` for polling:
379 
380```swift
381struct PingService {
382    func startPinging() -> AsyncStream<Bool> {
383        AsyncStream {
384            try? await Task.sleep(for: .seconds(5))
385            return await ping()
386        } onCancel: {
387            print("Pinging cancelled")
388        }
389    }
390    
391    func ping() async -> Bool {
392        // Network request
393        return true
394    }
395}
396 
397// Usage
398for await result in pingService.startPinging() {
399    print("Ping: \(result)")
400}
401```
402 
403## Standard Library Integration
404 
405### NotificationCenter
406 
407```swift
408let stream = NotificationCenter.default.notifications(
409    named: .NSSystemTimeZoneDidChange
410)
411 
412for await notification in stream {
413    print("Time zone changed")
414}
415```
416 
417### Combine publishers
418 
419```swift
420let numbers = [1, 2, 3, 4, 5]
421let filtered = numbers.publisher.filter { $0 % 2 == 0 }
422 
423for await number in filtered.values {
424    print(number) // 2, 4
425}
426```
427 
428### Task groups
429 
430```swift
431await withTaskGroup(of: Image.self) { group in
432    for url in urls {
433        group.addTask { await download(url) }
434    }
435    
436    for await image in group {
437        display(image)
438    }
439}
440```
441 
442## Limitations
443 
444### Single consumer only
445 
446Unlike Combine, streams support one consumer at a time:
447 
448```swift
449let stream = AsyncStream { continuation in
450    (0...5).forEach { continuation.yield($0) }
451    continuation.finish()
452}
453 
454Task {
455    for await value in stream {
456        print("Consumer 1: \(value)")
457    }
458}
459 
460Task {
461    for await value in stream {
462        print("Consumer 2: \(value)")
463    }
464}
465 
466// Unpredictable output - values split between consumers
467// Consumer 1: 0
468// Consumer 2: 1
469// Consumer 1: 2
470// Consumer 2: 3
471```
472 
473**Solution**: Create separate streams or use third-party libraries (AsyncExtensions).
474 
475### No values after termination
476 
477Once finished, stream won't emit new values:
478 
479```swift
480let stream = AsyncStream<Int> { continuation in
481    continuation.finish() // Terminate immediately
482    continuation.yield(1) // Never received
483}
484 
485for await value in stream {
486    print(value) // Loop exits immediately
487}
488```
489 
490## Decision Guide
491 
492### Use AsyncSequence when:
493 
494- Implementing standard library-style protocols
495- Need fine-grained control over iteration
496- Building reusable sequence types
497- Working with existing sequence protocols
498 
499**Reality**: Rarely needed in application code.
500 
501### Use AsyncStream when:
502 
503- Bridging delegates to async/await
504- Converting closure-based APIs
505- Emitting events manually
506- Polling or repeated async operations
507- Most common use case
508 
509---
510 
511## When to Use AsyncAlgorithms vs Standard Library
512 
513### Use AsyncAlgorithms when:
514 
515- **Time-based operations** need debounce/throttle/timer
516- **Combining multiple async sequences** (merge, combineLatest, zip)
517- **Multi-consumer scenarios** require backpressure (AsyncChannel)
518- **Complex operator chains** that Combine would handle naturally
519- **Need specific operators** not in standard library
520 
521### Use Standard Library when:
522 
523- **Bridging callback APIs** → AsyncStream
524- **Simple iteration** → for await in sequence
525- **Single-value operations** → async/await
526- **Basic transformations** → map/filter/contains
527 
528### Quick Decision Table
529 
530| Need | Solution |
531|------|----------|
532| Debounce search input | ✅ AsyncAlgorithms.debounce() |
533| Throttle button clicks | ✅ AsyncAlgorithms.throttle() |
534| Merge independent streams | ✅ AsyncAlgorithms.merge() |
535| Combine dependent values | ✅ AsyncAlgorithms.combineLatest() or async let |
536| Pair values from two sources | ✅ AsyncAlgorithms.zip() |
537| Bridge callback API | AsyncStream |
538| Multi-consumer with backpressure | ✅ AsyncChannel |
539| Periodic timer | ✅ AsyncTimerSequence |
540| Simple async iteration | for await in... |
541 
542> **See**: [async-algorithms.md](async-algorithms.md) for detailed usage examples with real-world patterns.
543 
544### Use regular async methods when:
545 
546- Single value returned
547- No progress updates needed
548- Simple request/response pattern
549 
550```swift
551// Use this
552func fetchData() async throws -> Data
553 
554// Not this
555func fetchData() -> AsyncThrowingStream<Data, Error>
556 
557> **Course Deep Dive**: This topic is covered in detail in [Lesson 6.3: Deciding between AsyncSequence, AsyncStream, or regular asynchronous methods](https://www.swiftconcurrencycourse.com?utm_source=github&utm_medium=agent-skill&utm_campaign=lesson-reference)
558```
559 
560## Common Patterns
561 
562### Progress reporting
563 
564```swift
565func download(_ url: URL) -> AsyncThrowingStream<DownloadEvent, Error> {
566    AsyncThrowingStream { continuation in
567        Task {
568            do {
569                var progress: Double = 0
570                while progress < 1.0 {
571                    progress += 0.1
572                    continuation.yield(.progress(progress))
573                    try await Task.sleep(for: .milliseconds(100))
574                }
575                
576                let data = try await URLSession.shared.data(from: url).0
577                continuation.yield(.completed(data))
578                continuation.finish()
579            } catch {
580                continuation.finish(throwing: error)
581            }
582        }
583    }
584}
585```
586 
587### Monitoring file system
588 
589```swift
590func watchDirectory(_ path: String) -> AsyncStream<FileEvent> {
591    AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in
592        let source = DispatchSource.makeFileSystemObjectSource(
593            fileDescriptor: fd,
594            eventMask: .write,
595            queue: .main
596        )
597        
598        source.setEventHandler {
599            continuation.yield(.fileChanged(path))
600        }
601        
602        continuation.onTermination = { _ in
603            source.cancel()
604        }
605        
606        source.resume()
607    }
608}
609```
610 
611### Timer/polling
612 
613```swift
614func timer(interval: Duration) -> AsyncStream<Date> {
615    AsyncStream { continuation in
616        Task {
617            while !Task.isCancelled {
618                continuation.yield(Date())
619                try? await Task.sleep(for: interval)
620            }
621            continuation.finish()
622        }
623    }
624}
625 
626// Usage
627for await date in timer(interval: .seconds(1)) {
628    print("Tick: \(date)")
629}
630```
631 
632## Best Practices
633 
6341. **Always call finish()** - Streams stay alive until terminated
6352. **Use buffer policies wisely** - Match your use case (latest value vs all values)
6363. **Handle cancellation** - Set `onTermination` for cleanup
6374. **Single consumer** - Don't share streams across multiple consumers
6385. **Prefer streams over closures** - More composable and cancellable
6396. **Check Task.isCancelled** - Respect cancellation in custom sequences
6407. **Use throwing variant** - When operations can fail
6418. **Consider regular async** - If only returning single value
642 
643## Debugging
644 
645### Add termination logging
646 
647```swift
648continuation.onTermination = { reason in
649    print("Stream ended: \(reason)")
650}
651```
652 
653### Validate finish() calls
654 
655```swift
656// ❌ Forgot to finish
657AsyncStream { continuation in
658    continuation.yield(1)
659    // Stream never ends!
660}
661 
662// ✅ Always finish
663AsyncStream { continuation in
664    continuation.yield(1)
665    continuation.finish()
666}
667```
668 
669### Check for dropped values
670 
671```swift
672let stream = AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in
673    for i in 1...100 {
674        continuation.yield(i)
675        print("Yielded: \(i)")
676    }
677    continuation.finish()
678}
679 
680// If consumer is slow, many values dropped
681for await value in stream {
682    print("Received: \(value)")
683    try? await Task.sleep(for: .seconds(1))
684}
685```
686 
687## Common Mistakes Agents Make
688 
689```swift
690// ❌ Values after finish() are silently dropped
691continuation.finish()
692continuation.yield(1) // Never received
693 
694// ❌ Stream never terminates (forgot finish)
695AsyncStream { continuation in
696    continuation.yield(1)
697    // Missing: continuation.finish()
698}
699 
700// ❌ Wrapping a single-value API in a stream — use a regular async function instead
701func fetchUser() -> AsyncStream<User> { ... } // Overkill for one result
702```
703 
704- **Sharing a single `AsyncStream` between multiple consumers**: Values split unpredictably. There is no built-in broadcast; use `AsyncChannel` for point-to-point multi-consumer patterns.
705- **Forgetting `onTermination`** when bridging delegate or observer APIs, causing resources to leak.
706 
707## Further Learning
708 
709For real-world migration examples, performance patterns, and advanced stream techniques, see [Swift Concurrency Course](https://www.swiftconcurrencycourse.com).
710 
711