Creating an Observable around an async/await method using RxSwift

Creating an Observable around an async/await method using RxSwift

Problem Description:

I am using the AWS Amplify library https://github.com/aws-amplify/amplify-swift to communicate with the Cognito service. Most of the functions have been rewritten using the new async/await approach.

Looking at the following method:

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

How can I wrap the await call to return an Observable<AuthSession> using RxSwift?

Solution – 1

Use Observable.create to create an Observable.

Use Task { ... } to perform async work.

Use Task { ... } inside Observable.create to perform async work in an Observable.

Something like this should work:

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()
    }
}

Solution – 2

It might help to see a generic version:

extension Observable {
    static func create(_ fn: @escaping () async throws -> Element) -> Observable<Element> {
        Observable.create { observer in
            let task = Task {
                do {
                    observer.on(.next(try await fn()))
                    observer.on(.completed)
                } catch {
                    observer.on(.error(error))
                }
            }
            return Disposables.create { task.cancel() }
        }
    }
}

Which would be used in this context like this:

func fetchAuthSession() -> Observable<AuthSession> {
    .create(Amplify.Auth.fetchAuthSession)
}