1# AsyncAlgorithms Package
2 
3Use this when:
4 
5- You need time-based operators (debounce, throttle, timers).
6- You need to combine multiple async sequences (merge, combineLatest, zip).
7- You are migrating from Combine or RxSwift operators to Swift Concurrency equivalents.
8 
9Skip this file if:
10 
11- You need basic `AsyncStream` bridging for callbacks or delegates. Use `async-sequences.md`.
12- You are choosing between `Task`, `async let`, or task groups. Use `tasks.md`.
13 
14Jump to:
15 
16- Quick Start
17- Time-Based Operators
18- Combining Operators
19- Multi-Consumer Scenarios
20- Combine Migration Guide
21- Best Practices
22 
23---
24 
25## Quick Start
26 
27Top 5 most common operators:
28 
29```swift
30import AsyncAlgorithms
31 
32// 1. Debounce rapid inputs
33for await query in searchQueryStream.debounce(for: .milliseconds(500)) {
34    await performSearch(query)
35}
36 
37// 2. Throttle repeated actions
38for await _ in buttonClicks.throttle(for: .seconds(1)) {
39    await performAction()
40}
41 
42// 3. Merge multiple independent streams
43for await message in chat1Messages.merge(chat2Messages) {
44    display(message)
45}
46 
47// 4. Combine dependent values
48for await (username, email) in usernameStream.combineLatest(emailStream) {
49    validateForm(username: username, email: email)
50}
51 
52// 5. Zip paired operations
53for await (image, metadata) in imageStream.zip(metadataStream) {
54    await cache(image: image, metadata: metadata)
55}
56```
57 
58> **See**: [AsyncAlgorithms on GitHub](https://github.com/apple/swift-async-algorithms)
59 
60---
61 
62## Overview & Installation
63 
64### What is AsyncAlgorithms?
65 
66Extends Swift's AsyncSequence with time-based operators, stream combination tools, and multi-consumer primitives.
67 
68**Use for**:
69- Time-based operations: debounce, throttle, timers
70- Combining streams: merge, combineLatest, zip, chain
71- Multi-consumer scenarios: AsyncChannel for backpressure
72- Specific operators: removeDuplicates, chunks, adjacentPairs, compacted
73 
74**Use standard library for**:
75- Bridging callbacks: AsyncStream
76- Simple iteration: for await in sequence
77- Single-value operations: async/await
78 
79### Installation
80 
81```swift
82dependencies: [
83    .package(url: "https://github.com/apple/swift-async-algorithms", from: "1.0.0")
84]
85 
86targets: [
87    .target(
88        name: "MyTarget",
89        dependencies: [
90            .product(name: "AsyncAlgorithms", package: "swift-async-algorithms")
91        ]
92    )
93]
94```
95 
96Import:
97 
98```swift
99import AsyncAlgorithms
100```
101 
102---
103 
104## Time-Based Operators
105 
106### debounce(for:tolerance:clock:)
107 
108Wait for inactivity before emitting. Use for rapid inputs like search fields.
109 
110#### Example: ArticleSearcher
111 
112```swift
113import AsyncAlgorithms
114 
115@Observable
116final class ArticleSearcher {
117    @MainActor private(set) var results: [Article] = []
118    private var searchQueryContinuation: AsyncStream<String>.Continuation?
119 
120    private lazy var searchQueryStream: AsyncStream<String> = {
121        AsyncStream { continuation in
122            searchQueryContinuation = continuation
123        }
124    }()
125 
126    func search(_ query: String) {
127        searchQueryContinuation?.yield(query)
128    }
129 
130    func startDebouncedSearch() {
131        Task { @MainActor in
132            for await query in searchQueryStream.debounce(for: .milliseconds(500)) {
133                self.results = []
134                self.results = await APIClient.searchArticles(query)
135            }
136        }
137    }
138}
139```
140 
141**Benefits**: Automatic cancellation, backpressure, cleaner than manual Task.sleep.
142 
143#### ❌ Anti-Pattern
144 
145```swift
146// Bad: Every keystroke spawns new task
147func search(_ query: String) {
148    Task {
149        try? await Task.sleep(for: .milliseconds(500))
150        await performSearch(query)
151    }
152}
153```
154 
155**Problem**: Multiple tasks execute simultaneously, causing out-of-order results.
156 
157**Solution**: Use `debounce()` for automatic backpressure.
158 
159---
160 
161### throttle(for:clock:reducing:)
162 
163Emit at most one value per interval. Use for repeated actions like button taps.
164 
165#### Example: Like Button
166 
167```swift
168import AsyncAlgorithms
169 
170struct LikeButton: View {
171    @State private var tapStream = AsyncStream<Void> { continuation in
172        // Continuation stored externally
173    }
174    @State private var isLiked = false
175 
176    var body: some View {
177        Button(action: {
178            tapStream.continuation?.yield()
179        }) {
180            Image(systemName: isLiked ? "heart.fill" : "heart")
181        }
182        .task {
183            await handleThrottledTaps()
184        }
185    }
186 
187    private func handleThrottledTaps() async {
188        for await _ in tapStream.throttle(for: .seconds(1)) {
189            await toggleLike()
190        }
191    }
192 
193    private func toggleLike() async {
194        isLiked.toggle()
195        await APIClient.updateLikeStatus(isLiked: isLiked)
196    }
197}
198```
199 
200#### Understanding reducing Parameter
201 
202```swift
203// .latest (default): Keep most recent value
204for await value in events.throttle(for: .seconds(1)) {
205    process(value)
206}
207 
208// .oldest: Keep first value
209for await value in events.throttle(for: .seconds(1), reducing: .oldest) {
210    process(value)
211}
212 
213// Custom: Sum all values
214for await value in events.throttle(for: .seconds(1)) { $0 + $1 } {
215    process(value)
216}
217```
218 
219---
220 
221### AsyncTimerSequence
222 
223Emit values at regular intervals. Use for periodic refresh or countdown timers.
224 
225#### Example: Feed Refresh
226 
227```swift
228import AsyncAlgorithms
229 
230@MainActor @Observable
231final class FeedViewModel {
232    private(set) var articles: [Article] = []
233    private var refreshTask: Task<Void, Never>?
234 
235    func startAutoRefresh() {
236        refreshTask = Task {
237            for await _ in AsyncTimerSequence(interval: .seconds(30)) {
238                await refreshFeed()
239            }
240        }
241    }
242 
243    private func refreshFeed() async {
244        articles = await APIClient.fetchLatestArticles()
245    }
246}
247```
248 
249#### ❌ Anti-Pattern
250 
251```swift
252// Bad: Manual timer implementation
253func startTimer() {
254    Task {
255        while !Task.isCancelled {
256            performAction()
257            try? await Task.sleep(for: .seconds(1))
258        }
259    }
260}
261```
262 
263**Solution**: Use `AsyncTimerSequence`.
264 
265---
266 
267## Combining Operators
268 
269### merge(_:...)
270 
271Combine sequences into one, emitting as they arrive. **Stable operator ✅**
272 
273Use for independent data sources that don't depend on each other.
274 
275#### Example: Multi-Room Chat
276 
277```swift
278import AsyncAlgorithms
279 
280actor ChatManager {
281    private var messageContinuations: [String: AsyncStream<ChatMessage>.Continuation] = [:]
282 
283    func getMessagesStream(roomID: String) -> AsyncStream<ChatMessage> {
284        AsyncStream { continuation in
285            messageContinuations[roomID] = continuation
286        }
287    }
288 
289    func receiveMessage(_ message: ChatMessage) {
290        messageContinuations[message.roomID]?.yield(message)
291    }
292 
293    func startMonitoring(rooms: [String]) -> AsyncStream<ChatMessage> {
294        let streams = rooms.map { getMessagesStream(roomID: $0) }
295        return streams.merge()
296    }
297}
298 
299// Usage
300let manager = ChatManager()
301let mergedMessages = await manager.startMonitoring(rooms: ["general", "random"])
302 
303for await message in mergedMessages {
304    print("[\(message.roomID)] \(message.text)")
305}
306```
307 
308**Behavior**: Values emit as they arrive from any source. Order interleaved by timing. Cancellation propagates to all sources.
309 
310---
311 
312### combineLatest(_:...)
313 
314Combine sequences, emitting tuple when any source emits. Always uses latest values. **Stable operator ✅**
315 
316Use for dependent values that need synchronization.
317 
318#### Example: Form Validation
319 
320```swift
321import AsyncAlgorithms
322 
323struct SignupForm: View {
324    @State private var usernameStream = AsyncStream<String> { /* ... */ }
325    @State private var emailStream = AsyncStream<String> { /* ... */ }
326    @State private var passwordStream = AsyncStream<String> { /* ... */ }
327    @State private var formState = FormState.incomplete
328 
329    var body: some View {
330        Form {
331            TextField("Username", text: $username)
332            TextField("Email", text: $email)
333            SecureField("Password", text: $password)
334        }
335        .task {
336            await validateForm()
337        }
338    }
339 
340    private func validateForm() async {
341        for await (username, email, password) in
342                usernameStream.combineLatest(emailStream, passwordStream)
343        {
344            formState = await validate(
345                username: username,
346                email: email,
347                password: password
348            )
349        }
350    }
351}
352```
353 
354#### ❌ Anti-Pattern
355 
356```swift
357// Bad: Manual value combining
358actor FormValidator {
359    private var currentUsername: String = ""
360    private var currentEmail: String = ""
361 
362    func updateUsername(_ username: String) {
363        currentUsername = username
364        checkForm()
365    }
366}
367```
368 
369**Solution**: Use `combineLatest()`.
370 
371---
372 
373### zip(_:...)
374 
375Combine sequences by pairing elements in order. **Stable operator ✅**
376 
377#### Example: Image + Metadata
378 
379```swift
380import AsyncAlgorithms
381 
382struct ImageLoader {
383    func loadImagesWithMetadata(urls: [URL]) async throws -> [LoadedImage] {
384        let imageStream = AsyncThrowingStream<UIImage, Error> { continuation in
385            Task {
386                for url in urls {
387                    let image = try await downloadImage(from: url)
388                    continuation.yield(image)
389                }
390                continuation.finish()
391            }
392        }
393 
394        let metadataStream = AsyncThrowingStream<ImageMetadata, Error> { continuation in
395            Task {
396                for url in urls {
397                    let metadata = try await fetchMetadata(for: url)
398                    continuation.yield(metadata)
399                }
400                continuation.finish()
401            }
402        }
403 
404        var results: [LoadedImage] = []
405        for try await (image, metadata) in imageStream.zip(metadataStream) {
406            results.append(LoadedImage(image: image, metadata: metadata))
407        }
408        return results
409    }
410}
411```
412 
413**Behavior**: Emits tuple when all sequences emit. Maintains order. Finishes when shortest sequence finishes.
414 
415---
416 
417### chain(_:...)
418 
419Concatenate sequences sequentially. **Stable operator ✅**
420 
421#### Example: Paginated Loading
422 
423```swift
424import AsyncAlgorithms
425 
426struct ArticlePaginator {
427    func loadAllArticles() -> AsyncStream<[Article]> {
428        AsyncStream { continuation in
429            Task {
430                var page = 1
431                var hasMore = true
432                while hasMore {
433                    let articles = try await fetchPage(page: page)
434                    continuation.yield(articles)
435                    hasMore = articles.count == 20
436                    page += 1
437                }
438                continuation.finish()
439            }
440        }
441    }
442}
443 
444// Usage: Chain cache + network
445for await articles in loadFromCacheStream().chain(loadFromNetworkStream()) {
446    display(articles)
447}
448```
449 
450**Behavior**: Emits all values from first sequence before starting second.
451 
452---
453 
454## Utility Operators
455 
456### removeDuplicates()
457 
458Remove adjacent duplicates. **Stable operator ✅**
459 
460```swift
461import AsyncAlgorithms
462 
463actor ChatHistory {
464    private var messageStream = AsyncStream<ChatMessage> { /* ... */ }
465 
466    func getUniqueMessages() -> AsyncStream<ChatMessage> {
467        messageStream.removeDuplicates()
468    }
469}
470```
471 
472---
473 
474### chunks() and chunked()
475 
476Collect values into batches. **Stable operator ✅**
477 
478```swift
479import AsyncAlgorithms
480 
481struct BatchProcessor {
482    func processLargeDataset(dataStream: AsyncStream<DataItem>) async {
483        for await batch in dataStream.chunks(count: 100) {
484            await processBatch(batch)
485        }
486    }
487 
488    func chunkedByTime(dataStream: AsyncStream<DataItem>) async {
489        for await batch in dataStream.chunked(by: .seconds(5)) {
490            await processBatch(batch)
491        }
492    }
493}
494```
495 
496---
497 
498### compacted() and adjacentPairs()
499 
500```swift
501import AsyncAlgorithms
502 
503// Remove nil values
504for await value in optionalValuesStream.compacted() {
505    process(value)
506}
507 
508// Pair adjacent elements
509for await (previous, current) in valuesStream.adjacentPairs() {
510    let difference = current - previous
511}
512```
513 
514---
515 
516## Multi-Consumer Scenarios
517 
518### AsyncChannel
519 
520AsyncSequence with backpressure. **Stable operator ✅**
521 
522Use for producer-consumer patterns with flow control.
523 
524#### Example: Message Queue
525 
526```swift
527import AsyncAlgorithms
528 
529actor MessageQueue {
530    private let channel = AsyncChannel<Message>()
531 
532    func getMessages() -> AsyncStream<Message> {
533        channel
534    }
535 
536    func enqueue(_ message: Message) async {
537        await channel.send(message)
538    }
539 
540    func startProcessing() {
541        Task {
542            for await message in channel {
543                await process(message)
544            }
545        }
546    }
547}
548 
549// Multiple producers
550let queue = MessageQueue()
551Task { await queue.enqueue(Message(type: .userAction, content: "tap")) }
552Task { await queue.enqueue(Message(type: .network, content: "data")) }
553queue.startProcessing()
554```
555 
556#### ❌ Anti-Pattern
557 
558```swift
559// Bad: Values split unpredictably
560let stream = AsyncStream<Int> { continuation in
561    for i in 1...10 {
562        continuation.yield(i)
563    }
564    continuation.finish()
565}
566 
567Task { for await value in stream { print("Consumer 1: \(value)") } }
568Task { for await value in stream { print("Consumer 2: \(value)") } }
569```
570 
571**Problem**: Each value goes to only one consumer.
572 
573**Solution**: Use `AsyncChannel` for multi-consumer scenarios.
574 
575---
576 
577### AsyncThrowingChannel
578 
579Like AsyncChannel but can emit errors. **Stable operator ✅**
580 
581#### Example: WebSocket
582 
583```swift
584import AsyncAlgorithms
585 
586actor WebSocketConnection {
587    private let channel = AsyncThrowingChannel<WebSocketMessage, Error>()
588 
589    func getMessages() -> AsyncThrowingStream<WebSocketMessage, Error> {
590        channel
591    }
592 
593    func receiveMessage(_ message: WebSocketMessage) async {
594        await channel.send(message)
595    }
596 
597    func reportError(_ error: Error) async {
598        await channel.finish(throwing: error)
599    }
600}
601 
602// Usage
603do {
604    for await message in connection.getMessages() {
605        handle(message)
606    }
607} catch {
608    print("WebSocket error: \(error)")
609}
610```
611 
612---
613 
614## Combine Migration Guide
615 
616### Operator Mapping Table
617 
618| Combine | AsyncAlgorithms | Status | Alternative |
619|---------|-----------------|---------|-------------|
620| `.debounce()` | `debounce()` | ✅ Stable | - |
621| `.throttle()` | `throttle()` | ✅ Stable | - |
622| `.merge()` | `merge()` | ✅ Stable | - |
623| `.combineLatest()` | `combineLatest()` | ✅ Stable | - |
624| `.zip()` | `zip()` | ✅ Stable | - |
625| `.concat()` | `chain()` | ✅ Stable | - |
626| `.removeDuplicates()` | `removeDuplicates()` | ✅ Stable | - |
627| `.timer()` | `AsyncTimerSequence` | ✅ Stable | - |
628| `.share()` | - | - | `AsyncChannel` |
629| `.flatMap()` | - | - | `TaskGroup` |
630| `.receive(on:)` | - | - | `Task` / `@MainActor` |
631| `.eraseToAnyPublisher()` | - | - | `any AsyncSequence` |
632 
633---
634 
635### Migration Examples
636 
637#### Example 1: ArticleSearcher
638 
639**Before: Combine**
640 
641```swift
642import Combine
643 
644final class ArticleSearcher: ObservableObject {
645    @Published private(set) var results: [Article] = []
646    @Published var searchQuery = ""
647 
648    init() {
649        $searchQuery
650            .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main)
651            .removeDuplicates()
652            .flatMap { query in
653                APIClient.searchArticles(query)
654                    .catch { _ in Just([]) }
655            }
656            .receive(on: DispatchQueue.main)
657            .assign(to: &$results)
658    }
659}
660```
661 
662**After: AsyncAlgorithms**
663 
664```swift
665import AsyncAlgorithms
666 
667@Observable
668final class ArticleSearcher {
669    @MainActor private(set) var results: [Article] = []
670    private var searchQueryContinuation: AsyncStream<String>.Continuation?
671 
672    private lazy var searchQueryStream: AsyncStream<String> = {
673        AsyncStream { continuation in
674            searchQueryContinuation = continuation
675        }
676    }()
677 
678    func search(_ query: String) {
679        searchQueryContinuation?.yield(query)
680    }
681 
682    func startDebouncedSearch() {
683        Task { @MainActor in
684            for await query in searchQueryStream
685                .debounce(for: .milliseconds(500))
686                .removeDuplicates()
687            {
688                do {
689                    self.results = try await APIClient.searchArticles(query)
690                } catch {
691                    self.results = []
692                }
693            }
694        }
695    }
696}
697```
698 
699**Benefits**: Simpler error handling, no cancellables, automatic cancellation.
700 
701---
702 
703#### Example 2: Multi-Source Loading
704 
705**Before: Combine Merge**
706 
707```swift
708import Combine
709 
710final class ArticleLoader: ObservableObject {
711    @Published private(set) var items: [Item] = []
712 
713    func loadAllSources() {
714        let source1 = APIClient.fetchItems(from: .source1)
715        let source2 = APIClient.fetchItems(from: .source2)
716 
717        Publishers.Merge(source1, source2)
718            .scan([]) { accumulated, new in
719                accumulated + new
720            }
721            .receive(on: DispatchQueue.main)
722            .assign(to: &$items)
723    }
724}
725```
726 
727**After: TaskGroup**
728 
729```swift
730import AsyncAlgorithms
731 
732@Observable
733final class ArticleLoader {
734    @MainActor private(set) var items: [Item] = []
735 
736    func loadAllSourcesParallel() async {
737        await withTaskGroup(of: [Item].self) { group in
738            group.addTask {
739                await APIClient.fetchItems(from: .source1)
740            }
741            group.addTask {
742                await APIClient.fetchItems(from: .source2)
743            }
744 
745            for await newItems in group {
746                items.append(contentsOf: newItems)
747            }
748        }
749    }
750}
751```
752 
753**Key difference**: For parallel execution, use `TaskGroup` instead of `flatMap`.
754 
755---
756 
757#### Example 3: Form Validation
758 
759**Before: Combine**
760 
761```swift
762import Combine
763 
764final class FormValidator: ObservableObject {
765    @Published var username = ""
766    @Published var email = ""
767 
768    @Published private(set) var formState: FormState = .incomplete
769 
770    init() {
771        Publishers.CombineLatest2($username, $email)
772            .map { username, email in
773                validate(username: username, email: email)
774            }
775            .assign(to: &$formState)
776    }
777}
778```
779 
780**After: AsyncAlgorithms or async let**
781 
782```swift
783import AsyncAlgorithms
784 
785@Observable
786final class FormValidator {
787    var username = ""
788    var email = ""
789 
790    @MainActor private(set) var formState: FormState = .incomplete
791 
792    // Option 1: combineLatest for stream-based validation
793    func startStreamValidation() {
794        Task { @MainActor in
795            for await (username, email) in
796                    usernameStream.combineLatest(emailStream)
797            {
798                self.formState = validate(
799                    username: username,
800                    email: email
801                )
802            }
803        }
804    }
805 
806    // Option 2: async let for simple validation
807    func validateForm() async {
808        let (username, email) = await (username, email)
809        formState = validate(
810            username: username,
811            email: email
812        )
813    }
814}
815```
816 
817**Choose**:
818- `combineLatest()`: Continuous validation as fields change
819- `async let`: One-time validation when all values available
820 
821---
822 
823## Common Mistakes Agents Make
824 
825- **Manual debounce with `Task.sleep`**: This creates multiple concurrent tasks and risks out-of-order results. Use the stream-based `debounce(for:)` operator from AsyncAlgorithms instead.
826- **Sharing `AsyncStream` across multiple consumers**: Values split unpredictably between consumers. Use `AsyncChannel` for multi-consumer scenarios with backpressure. Note: `AsyncChannel` is point-to-point, not broadcast like Combine's `.share()`.
827- **Looking for a `.flatMap` equivalent**: Use `TaskGroup` for fan-out; the semantics differ from Combine/Rx `flatMap`.
828- **Looking for `.receive(on:)` equivalent**: Use `@MainActor` or `Task` context for isolation instead.
829 
830## Best Practices
831 
8321. **Use time-based operators** for rapid inputs: debounce() for search, throttle() for buttons
8332. **Combine streams** with merge/combineLatest instead of manual state management
8343. **Use AsyncChannel** for multi-consumer scenarios with backpressure
8354. **Ensure Sendable conformance** when using operators across isolation boundaries
8365. **Leverage cancellation** - Task cancellation propagates through all operators
8376. **Choose right tool**: AsyncAlgorithms for complex streams, AsyncStream for bridging callbacks
8387. **Avoid manual sleep loops** - use AsyncTimerSequence instead
839 
840---
841 
842## Further Learning
843 
844- [AsyncAlgorithms Documentation](https://github.com/apple/swift-async-algorithms)
845- [Combine Migration Guide](migration.md)
846- [Async Sequences](async-sequences.md)
847- [Tasks](tasks.md) - Task groups and structured concurrency
848