Rork Max × Swift Concurrency: The Complete Implementation Guide — Mastering async/await, Actors, and Structured Concurrency

// Dangerous: shared mutable state without synchronization
class UserCache {
    var cache: [String: User] = [:]  // ⚠️ Not thread-safe
 
    func store(_ user: User) {
        cache[user.id] = user  // Race condition if called from multiple threads
    }
 
    func retrieve(id: String) -> User? {
        return cache[id]  // Simultaneous read/write can crash
    }
}

// Thread-safe cache using Actor
actor UserCache {
    // All access to Actor properties is automatically serialized
    private var cache: [String: User] = [:]
    private var lastAccessTime: [String: Date] = [:]
 
    // Actor methods are implicitly async from outside
    func store(_ user: User) {
        cache[user.id] = user
        lastAccessTime[user.id] = Date()
    }
 
    func retrieve(id: String) -> User? {
        return cache[id]
    }
 
    // Prune entries older than a given threshold
    func pruneExpiredEntries(olderThan threshold: TimeInterval) {
        let cutoff = Date().addingTimeInterval(-threshold)
        let expiredKeys = lastAccessTime
            .filter { $0.value < cutoff }
            .map { $0.key }
 
        expiredKeys.forEach { key in
            cache.removeValue(forKey: key)
            lastAccessTime.removeValue(forKey: key)
        }
    }
}
 
// Usage
let userCache = UserCache()
 
Task {
    await userCache.store(User(id: "u1", name: "Masaki"))
    let user = await userCache.retrieve(id: "u1")
    print(user?.name ?? "not found")  // "Masaki"
}

// @MainActor ensures all access happens on the main thread
@MainActor
class ArticleViewModel: ObservableObject {
    @Published var articles: [Article] = []
    @Published var isLoading = false
    @Published var errorMessage: String?
 
    private let apiClient = ArticleAPIClient()
 
    func loadArticles() async {
        isLoading = true
        errorMessage = nil
 
        do {
            // ArticleAPIClient runs on a background thread
            let fetchedArticles = try await apiClient.fetchAll()
            // We're back on MainActor — safe to update @Published properties
            articles = fetchedArticles
        } catch {
            errorMessage = error.localizedDescription
        }
 
        isLoading = false
    }
}
 
// The API client doesn't need @MainActor — it runs in background
struct ArticleAPIClient {
    func fetchAll() async throws -> [Article] {
        let url = URL(string: "https://api.rorklab.net/articles")!
        let (data, _) = try await URLSession.shared.data(from: url)
        return try JSONDecoder().decode([Article].self, from: data)
    }
}

// Parallel data fetching with withThrowingTaskGroup
func fetchMultipleUsers(userIds: [String]) async throws -> [User] {
    try await withThrowingTaskGroup(of: User.self) { group in
        // Launch all fetches in parallel
        for userId in userIds {
            group.addTask {
                return try await UserAPI.fetchUser(userId: userId)
            }
        }
 
        // Collect results as they arrive
        var users: [User] = []
        for try await user in group {
            users.append(user)
        }
        return users
    }
    // By the time we reach this line, all child tasks have completed.
    // If any task throws, all others are automatically cancelled.
}

// Parallel dashboard data fetching in a Rork Max native module
actor DashboardDataManager {
    struct DashboardData {
        let user: User
        let recentArticles: [Article]
        let notifications: [Notification]
        let analytics: Analytics
    }
 
    func fetchDashboardData(userId: String) async throws -> DashboardData {
        // Start all four API calls immediately — they run in parallel
        async let user = UserAPI.fetchUser(userId: userId)
        async let articles = ArticleAPI.fetchRecent(limit: 10)
        async let notifications = NotificationAPI.fetchUnread(userId: userId)
        async let analytics = AnalyticsAPI.fetchSummary(userId: userId)
 
        // Wait for all four results simultaneously
        // If any one fails, the whole expression throws
        return try await DashboardData(
            user: user,
            recentArticles: articles,
            notifications: notifications,
            analytics: analytics
        )
        // Expected behavior: if each API takes ~500ms sequentially (~2000ms total),
        // parallel execution completes in ~500ms — a 4x speedup
    }
}

// Cancellable image download manager
class ImageDownloadManager {
    private var activeTasks: [String: Task<UIImage, Error>] = [:]
 
    func downloadImage(url: URL, cacheKey: String) async throws -> UIImage {
        // Cancel any existing download for this key
        activeTasks[cacheKey]?.cancel()
 
        let task = Task {
            // Cancellation checkpoint — throws CancellationError if cancelled
            try Task.checkCancellation()
 
            let (data, response) = try await URLSession.shared.data(from: url)
 
            // Check again after a potentially long operation
            try Task.checkCancellation()
 
            guard let httpResponse = response as? HTTPURLResponse,
                  httpResponse.statusCode == 200,
                  let image = UIImage(data: data) else {
                throw ImageError.invalidData
            }
 
            return image
        }
 
        activeTasks[cacheKey] = task
 
        do {
            let image = try await task.value
            activeTasks.removeValue(forKey: cacheKey)
            return image
        } catch {
            activeTasks.removeValue(forKey: cacheKey)
            throw error
        }
    }
 
    func cancelDownload(cacheKey: String) {
        activeTasks[cacheKey]?.cancel()
        activeTasks.removeValue(forKey: cacheKey)
    }
}

// WebSocket messages via AsyncStream
class WebSocketManager {
    private var webSocketTask: URLSessionWebSocketTask?
 
    func messageStream(url: URL) -> AsyncStream<String> {
        return AsyncStream { continuation in
            let session = URLSession(configuration: .default)
            let task = session.webSocketTask(with: url)
            self.webSocketTask = task
 
            func receiveNext() {
                task.receive { result in
                    switch result {
                    case .success(let message):
                        switch message {
                        case .string(let text):
                            continuation.yield(text)  // Send value to the stream
                        case .data(let data):
                            if let text = String(data: data, encoding: .utf8) {
                                continuation.yield(text)
                            }
                        @unknown default:
                            break
                        }
                        receiveNext()  // Continue receiving
                    case .failure:
                        continuation.finish()  // End the stream on error
                    }
                }
            }
 
            task.resume()
            receiveNext()
 
            // Clean up when the stream is cancelled (e.g., when the view disappears)
            continuation.onTermination = { _ in
                task.cancel(with: .goingAway, reason: nil)
            }
        }
    }
}
 
// SwiftUI integration
struct LiveMessageView: View {
    @StateObject private var viewModel = LiveMessageViewModel()
 
    var body: some View {
        List(viewModel.messages, id: \.self) { message in
            Text(message)
        }
        .task {
            // .task automatically cancels when the view disappears
            await viewModel.startReceiving()
        }
    }
}
 
@MainActor
class LiveMessageViewModel: ObservableObject {
    @Published var messages: [String] = []
    private let wsManager = WebSocketManager()
 
    func startReceiving() async {
        let url = URL(string: "wss://api.rorklab.net/ws/messages")!
        for await message in wsManager.messageStream(url: url) {
            messages.insert(message, at: 0)
            if messages.count > 100 {
                messages = Array(messages.prefix(100))
            }
        }
    }
}

// Convert any Combine Publisher to AsyncStream
extension Publisher where Failure == Never {
    var asyncStream: AsyncStream<Output> {
        AsyncStream { continuation in
            let cancellable = sink { _ in
                continuation.finish()
            } receiveValue: { value in
                continuation.yield(value)
            }
            continuation.onTermination = { _ in cancellable.cancel() }
        }
    }
}
 
// Example: battery level monitoring
func batteryLevelStream() -> AsyncStream<Float> {
    UIDevice.current.isBatteryMonitoringEnabled = true
    return NotificationCenter.default
        .publisher(for: UIDevice.batteryLevelDidChangeNotification)
        .compactMap { _ -> Float? in
            let level = UIDevice.current.batteryLevel
            return level >= 0 ? level : nil
        }
        .asyncStream
}
 
// Usage
Task {
    for await level in batteryLevelStream() {
        print("Battery: \(Int(level * 100))%")
        if level < 0.2 {
            await showLowBatteryAlert()
        }
    }
}

// ❌ Compiler warning: data race risk
class DataService {
    var data: [String] = []
 
    func process() async {
        await Task.detached {
            self.data.append("new item")  // Warning: non-Sendable type crossing isolation boundary
        }.value
    }
}
 
// ✅ Fix with Actor
actor DataService {
    var data: [String] = []
 
    func process() {
        data.append("new item")  // Safe inside Actor
    }
}

// ❌ Potential deadlock: synchronously waiting on the same Actor you're already on
@MainActor
class ViewModel {
    func badPattern() {
        // Synchronously waiting for a MainActor task while on MainActor can deadlock
        let _ = Task {
            await someMainActorWork()
        }
        // Never synchronously wait on task.value from within the same Actor
    }
 
    // ✅ Correct: expose as async method instead
    func goodPattern() async {
        await someMainActorWork()
    }
}

// ❌ Task runs forever even after the view controller is deallocated
class BadViewController: UIViewController {
    override func viewDidLoad() {
        super.viewDidLoad()
        Task {
            for await update in dataStream() {
                self.updateUI(update)  // self might be gone
            }
        }
    }
}
 
// ✅ Store the Task and cancel it on cleanup
class GoodViewController: UIViewController {
    private var streamTask: Task<Void, Never>?
 
    override func viewDidLoad() {
        super.viewDidLoad()
        streamTask = Task { [weak self] in
            guard let self else { return }
            for await update in dataStream() {
                await MainActor.run {
                    self.updateUI(update)
                }
            }
        }
    }
 
    override func viewDidDisappear(_ animated: Bool) {
        super.viewDidDisappear(animated)
        streamTask?.cancel()
        streamTask = nil
    }
}

// Production-ready Rork Max native module with Swift Concurrency
@objc(RorkDataModule)
class RorkDataModule: NSObject {
 
    @objc func fetchData(_ query: String,
                         resolve: @escaping RCTPromiseResolveBlock,
                         reject: @escaping RCTPromiseRejectBlock) {
        // Use Task.detached to avoid inheriting the caller's Actor context
        Task.detached(priority: .userInitiated) {
            do {
                let result = try await DataService.shared.search(query: query)
                await MainActor.run {
                    resolve(result)
                }
            } catch {
                await MainActor.run {
                    reject("SEARCH_ERROR", error.localizedDescription, error)
                }
            }
        }
    }
}
 
// Core service implemented as Actor
actor DataService {
    static let shared = DataService()
    private var cache: [String: [SearchResult]] = [:]
 
    func search(query: String) async throws -> [[String: Any]] {
        if let cached = cache[query] {
            return cached.map { $0.toDictionary() }
        }
 
        let results = try await SearchAPI.search(query: query)
        cache[query] = results
        return results.map { $0.toDictionary() }
    }
}

// Simple async performance wrapper
func measureAsync<T>(_ label: String, operation: () async throws -> T) async throws -> T {
    let start = CFAbsoluteTimeGetCurrent()
    let result = try await operation()
    let elapsed = CFAbsoluteTimeGetCurrent() - start
    print("⏱ \(label): \(String(format: "%.3f", elapsed * 1000))ms")
    return result
}
 
// Usage
let dashboardData = try await measureAsync("Dashboard parallel fetch") {
    try await dashboardManager.fetchDashboardData(userId: currentUserId)
}
// Expected output: ⏱ Dashboard parallel fetch: 312.456ms
// (vs ~1200ms sequential — ~4x improvement from parallelism alone)

@MainActor
class FeedViewModel: ObservableObject {
    @Published var items: [FeedItem] = []
 
    func refresh() {
        // ❌ This Task inherits @MainActor, so the heavy JSON decode
        //    runs on the main thread and blocks scrolling.
        Task {
            let data = try? await api.fetchFeed()        // Network is fine (await yields)
            let decoded = heavyDecode(data)              // ⚠️ This pins the main thread
            items = decoded
        }
    }
 
    func refreshCorrect() {
        // ✅ Push CPU-bound work off to a detached task
        Task {
            let data = try? await api.fetchFeed()
            let decoded = await Task.detached(priority: .userInitiated) {
                heavyDecode(data)                        // Runs in the background
            }.value
            items = decoded                              // Back on @MainActor for UI
        }
    }
}

actor ImageLoader {
    private var cache: [URL: UIImage] = [:]
    private var inFlight: [URL: Task<UIImage, Error>] = [:]
 
    func image(for url: URL) async throws -> UIImage {
        if let cached = cache[url] { return cached }
        // ✅ Share the in-flight task to avoid duplicate fetches
        if let existing = inFlight[url] { return try await existing.value }
 
        let task = Task { try await download(url) }
        inFlight[url] = task
        defer { inFlight[url] = nil }
 
        let image = try await task.value
        cache[url] = image
        return image
    }
}