如何用NWPathMonitor+async/await实现联网后执行带超时的任务

功能要求 #

使用NWPathMonitor检测网络连接，当网络就绪时执行指定异步任务，需符合以下方法签名：

func performWhenNetworkIsAvailable<T: Sendable>(timeoutDuration: Duration,
                                                _ operation: @escaping @Sendable () async -> T) async -> Result<T, NetworkOperationError>

对应的错误枚举：

public enum NetworkOperationError: Error {
    case timeout
}

使用示例：

let result = await networkOperationManager.performWhenNetworkIsAvailable(timeoutDuration: .seconds(10)) {
    return await fetchDataFromAPI()
}        

switch result {
    case .success(let data) :
        print("Got data: \(data)")
    case .failure(let error):
        print("Error fetching data: \(error.localizedDescription)")
}

现有问题 #

尝试用TaskGroup实现，但超时无法解锁任务组，现有代码如下：

actor NetworkOperationManager {
    private let monitor = NWPathMonitor()
    private var isNetworkAvailable = false
    private var continuations: [CheckedContinuation<Void, Never>] = []
    
    init() {
        Task {
            print("Starting monitor")
            monitor.start(queue: DispatchQueue(label: "NetworkMonitor"))

            for await path in monitor {
                print("New update from monitor : \(path.status == .satisfied ? "is connected" : "not connected")")
                let isAvailable = path.status == .satisfied
                guard await isAvailable != isNetworkAvailable else {
                    return
                }
                await setNetworkStatus(isAvailable: isAvailable)
                await resumeAllContinuations()
            }
        }
    }
    
    func setNetworkStatus(isAvailable: Bool) {
        isNetworkAvailable = isAvailable
    }
    
    func addContinuation(_ continuation: CheckedContinuation<Void, Never>) {
        continuations.append(continuation)
    }
    
    private func resumeAllContinuations() {
        continuations.forEach { $0.resume() }
        continuations.removeAll()
    }
    
    func performWhenNetworkIsAvailable<T: Sendable>(timeoutDuration: Duration,
                                                    _ operation: @escaping @Sendable () async -> T) async -> Result<T, NetworkOperationError> {
        guard !isNetworkAvailable else {
            print("Internet connection already available, executing operation")
            return await .success(operation())
        }
        
        let result = await withTaskGroup(of: Bool.self) { group -> Bool in
            group.addTask { @MainActor in
                try? await Task.sleep(for: timeoutDuration)
                print("End of timeout")
                return false
            }
            
            group.addTask { @MainActor [weak self] in
                guard await self?.isNetworkAvailable == false else {
                    print("Internet connection already available")
                    return true
                }
                
                await withCheckedContinuation { [weak self] continuation in
                    Task {
                        guard await self?.isNetworkAvailable == false else {
                            print("Internet connection already available")
                            continuation.resume()
                            return
                        }
                        await self?.addContinuation(continuation)
                    }
                }
                
                return true
            }
            
            return await group.next() ?? false
        }
        
        print("Got group result")
        
        if result {
            print("Success")
            return await .success(operation())
        } else {
            print("Failure")
            return .failure(NetworkOperationError.timeout)
        }
    }
}

期望目标 #

找到符合指定方法签名、高效且线程安全的实现方式，同时保持请求的收发顺序。

内容的提问来源于stack exchange，提问作者Xys