使用 RxSwift 围绕 async/await 方法创建 Observable

Question

我正在使用 AWS Amplify 库https://github.com/aws-amplify/amplify-swift与 Cognito 服务进行通信。大多数函数已使用新的 async/await 方法重写。

看下面的方法：

func fetchAuthSession() async throws -> AuthSession {
    return try await Amplify.Auth.fetchAuthSession()
}

Observable<AuthSession>如何使用 RxSwift包装等待调用以返回？

Answer 1

用于Observable.create创建一个Observable.

用于Task { ... }执行async工作。

使用Task { ... }insideObservable.create在.async​Observable

像这样的东西应该有效：

let authSessionObservable: Observable<AuthSession> = Observable.create { observer in
    let task = Task {
        do {
            let session = try await Amplify.Auth.fetchAuthSession()
            observer.on(.next(session))
            observer.on(.completed)
        } catch {
            observer.on(.error(error))
        }
    }
    return Disposables.create {
        task.cancel()
    }
}

Answer 2

正如其他人所说，您需要将异步调用包装到自定义Observable. 但是，我建议不要使用类型擦除类，而是创建一个Single：

extension Single {
    static func fromAsync<T>(_ fn: @escaping () async throws -> T) -> Single<T> {
        .create { observer in
            let task = Task {
                do { try await observer(.success(fn())) }
                catch { observer(.failure(error))}
            }
            return Disposables.create { task.cancel() }
        }
    }
}

ASingle更适合，因为它模拟了一个仅发出一个元素的 observable，这首先意味着更好的语义、更好的意图传输，以及在创建和使用 observable 时增加了编译器支持。

用法：

let fetchAuthSession = Single.fromAsync { try await Amplify.Auth.fetchAuthSession() }

，或（函数式编程风格）：

let fetchAuthSession = Single.fromAsync(Amplify.Auth.fetchAuthSession)