Swift Structured Concurrency Deep Dive: TaskGroup, AsyncStream ve Daha Fazlası

1// STRUCTURED: Child task'lar parent scope icinde
2func fetchDashboard() async throws -> Dashboard {
3    // async let ile paralel calisma
4    async let profile = fetchProfile()
5    async let notifications = fetchNotifications()
6    async let feed = fetchFeed()
7 
8    // Tum sonuclar hazir olana kadar bekle
9    // Herhangi biri hata firlatirsa, digerlerini iptal et
10    return try await Dashboard(
11        profile: profile,
12        notifications: notifications,
13        feed: feed
14    )
15}
16// Fonksiyon bittiginde tum child task'lar kesinlikle bitmis olur
17 
18// UNSTRUCTURED: Task bagimsiz yasar
19func startBackgroundSync() {
20    Task {
21        // Bu task, cagiran fonksiyondan bagimsiz yasiyor
22        // Kim cancel edecek? Kim hatayi handle edecek?
23        await syncAllData()
24    }
25}

1// MARK: - Paralel Image Download
2func downloadImages(urls: [URL]) async throws -> [UIImage] {
3    try await withThrowingTaskGroup(of: (Int, UIImage).self) { group in
4        // Her URL icin bir task olustur
5        for (index, url) in urls.enumerated() {
6            group.addTask {
7                let (data, _) = try await URLSession.shared.data(from: url)
8                guard let image = UIImage(data: data) else {
9                    throw ImageError.invalidData
10                }
11                return (index, image)
12            }
13        }
14 
15        // Sonuclari topla (sirasiz gelir!)
16        var results: [(Int, UIImage)] = []
17        for try await result in group {
18            results.append(result)
19        }
20 
21        // Orijinal siraya gore duz
22        return results
23            .sorted { $0.0 < $1.0 }
24            .map { $0.1 }
25    }
26}

1// MARK: - Throttled Paralel Islem
2func processItems<T, R>(
3    _ items: [T],
4    maxConcurrency: Int = 4,
5    operation: @Sendable @escaping(T) async throws -> R
6) async throws -> [R] {
7    try await withThrowingTaskGroup(of: (Int, R).self) { group in
8        var results: [(Int, R)] = []
9        var nextIndex = 0
10 
11        // Ilk batch'i ekle
12        for _ in 0..<min(maxConcurrency, items.count) {
13            let index = nextIndex
14            group.addTask {
15                let result = try await operation(items[index])
16                return (index, result)
17            }
18            nextIndex += 1
19        }
20 
21        // Biri bitince bir sonrakini ekle (sliding window)
22        for try await result in group {
23            results.append(result)
24 
25            if nextIndex < items.count {
26                let index = nextIndex
27                group.addTask {
28                    let result = try await operation(items[index])
29                    return (index, result)
30                }
31                nextIndex += 1
32            }
33        }
34 
35        return results.sorted { $0.0 < $1.0 }.map { $0.1 }
36    }
37}
38 
39// Kullanim
40// let processedImages = try await processItems(
41//     imageURLs,
42//     maxConcurrency: 3
43// ) { url in
44//     try await downloadAndResize(url)
45// }

1// MARK: - Location Stream
2extension CLLocationManager {
3    var locationStream: AsyncStream<CLLocation> {
4        AsyncStream { continuation in
5            let delegate = LocationDelegate(continuation: continuation)
6            self.delegate = delegate
7 
8            continuation.onTermination = { _ in
9                // Temizlik
10                self.stopUpdatingLocation()
11            }
12 
13            self.startUpdatingLocation()
14        }
15    }
16}
17 
18// Delegate -> AsyncStream koprusu
19private final class LocationDelegate: NSObject, CLLocationManagerDelegate {
20    let continuation: AsyncStream<CLLocation>.Continuation
21 
22    init(continuation: AsyncStream<CLLocation>.Continuation) {
23        self.continuation = continuation
24    }
25 
26    func locationManager(_ manager: CLLocationManager, didUpdateLocations locations: [CLLocation]) {
27        for location in locations {
28            continuation.yield(location)
29        }
30    }
31 
32    func locationManager(_ manager: CLLocationManager, didFailWithError error: Error) {
33        continuation.finish()
34    }
35}
36 
37// Kullanim
38// for await location in locationManager.locationStream {
39//     updateMap(with: location)
40// }

1// MARK: - WebSocket Stream
2func createWebSocketStream(url: URL) -> AsyncThrowingStream<WebSocketMessage, Error> {
3    AsyncThrowingStream { continuation in
4        let task = URLSession.shared.webSocketTask(with: url)
5 
6        func receiveNext() {
7            task.receive { result in
8                switch result {
9                case .success(let message):
10                    switch message {
11                    case .string(let text):
12                        continuation.yield(.text(text))
13                    case .data(let data):
14                        continuation.yield(.binary(data))
15                    @unknown default:
16                        break
17                    }
18                    receiveNext() // Sonraki mesaji bekle
19 
20                case .failure(let error):
21                    continuation.finish(throwing: error)
22                }
23            }
24        }
25 
26        continuation.onTermination = { _ in
27            task.cancel(with: .goingAway, reason: nil)
28        }
29 
30        task.resume()
31        receiveNext()
32    }
33}
34 
35enum WebSocketMessage {
36    case text(String)
37    case binary(Data)
38}

1// MARK: - Continuation ile Callback Bridge
2func requestCameraPermission() async -> Bool {
3    await withCheckedContinuation { continuation in
4        AVCaptureDevice.requestAccess(for: .video) { granted in
5            continuation.resume(returning: granted)
6        }
7    }
8}
9 
10// Throwing version
11func fetchFromLegacySDK(id: String) async throws -> LegacyResult {
12    try await withCheckedThrowingContinuation { continuation in
13        LegacySDK.fetch(id: id) { result, error in
14            if let error = error {
15                continuation.resume(throwing: error)
16            } else if let result = result {
17                continuation.resume(returning: result)
18            } else {
19                continuation.resume(throwing: LegacyError.unknown)
20            }
21        }
22    }
23    // DIKKAT: Continuation MUTLAKA tam olarak bir kez resume edilmeli!
24    // Sifir kez = memory leak, iki kez = crash
25}

1// MARK: - Cancellation-Aware Code
2func processLargeDataset(_ items: [DataItem]) async throws -> [ProcessedItem] {
3    var results: [ProcessedItem] = []
4 
5    for item in items {
6        // Her iterasyonda iptal kontrolu
7        try Task.checkCancellation()
8 
9        let processed = try await process(item)
10        results.append(processed)
11 
12        // Alternatif: cooperative cancellation
13        if Task.isCancelled {
14            // Temizlik yap ve partial sonuc don
15            await cleanup()
16            return results // Partial result
17        }
18    }
19 
20    return results
21}
22 
23// withTaskCancellationHandler
24func downloadFile(url: URL) async throws -> Data {
25    let sessionTask = URLSession.shared.dataTask(with: url)
26 
27    return try await withTaskCancellationHandler {
28        try await withCheckedThrowingContinuation { continuation in
29            sessionTask.completionHandler = { data, response, error in
30                if let error = error {
31                    continuation.resume(throwing: error)
32                } else if let data = data {
33                    continuation.resume(returning: data)
34                }
35            }
36            sessionTask.resume()
37        }
38    } onCancel: {
39        sessionTask.cancel() // Task iptal edilirse network istegini de iptal et
40    }
41}

1// MARK: - Timeout destekli TaskGroup wrapper
2func withTimeout<T: Sendable>(
3    seconds: TimeInterval,
4    operation: @Sendable @escaping() async throws -> T
5) async throws -> T {
6    try await withThrowingTaskGroup(of: T.self) { group in
7        // Ana islem
8        group.addTask { try await operation() }
9 
10        // Watchdog timer
11        group.addTask {
12            try await Task.sleep(nanoseconds: UInt64(seconds * 1_000_000_000))
13            throw TimeoutError(duration: seconds)
14        }
15 
16        // Ilk biten kazanir, digerini iptal et
17        guard let result = try await group.next() else {
18            throw TimeoutError(duration: seconds)
19        }
20        group.cancelAll()
21        return result
22    }
23}
24 
25struct TimeoutError: Error {
26    let duration: TimeInterval
27    var localizedDescription: String { "Islem \(duration)sn icinde tamamlanamadi" }
28}
29 
30// Kullanim: let data = try await withTimeout(seconds: 15) { try await fetchData() }