在另一个创建 rxswift 中订阅单个 observable

Question

我想从服务器获取数据并更新我的数据库，之后我会向用户显示收到的数据。为了这个目标，我的视图模型中有一个方法（getData()），它返回一个Single，我在视图控制器（myVC.getData.subscribe({single in ...})）中调用并订阅这个方法，首先我调用并订阅（#1） (getUnread()->Single) 方法运行但我无法获取单个事件，我不明白为什么我无法在回调(#4) 中获取事件(#3)

之后我想通过调用(#2)(save([Moddel])->single)来保存数据

//I removed some part of this code it was to big
//This method is View Model
func getData() -> Single<[Model]> {
    return Single<[Model]>.create {[weak self] single in
        //#1
        self!.restRepo.getUnread().subscribe({ [weak self] event in
            //#4
            switch event {
            case .success(let response):
                let models = response
                //#2
                self!.dbRepo.save(issues!).subscribe({ event in
                    switch event {
                    case .success(let response):
                        let models = response
                        single(.success(models))
                    case .error(let error):
                        single(.error(error))
                    }
                }).disposed(by: self!.disposeBag)
            case .error(let error):
                single(.error(error))
            }
        }).disposed(by: self!.disposeBag)
        return Disposables.create()
    }
}

。 .

 //I removed some part of this code it was to big
 //This method is in RestRepo class
 func getUnread() -> Single<[Model]> {
    
    return Single<[Model]>.create { single in
        let urlComponent = ApiHelper.instance.dolphinURLComponents(for: ApiHelper.ISSUES_PATH)
        var urlRequest = URLRequest(url: urlComponent.url!)

        ApiHelper.instance.alamofire.request(urlRequest).intercept().responseJSON { response in
            debugPrint(response)
            let statusCode = response.response?.statusCode
            switch statusCode {
            case 200:
                do {
                    let models = try JSONDecoder().decode([Model].self, from: response.data!)
                    //#3
                    single(.success(models))
                }catch{
                    print(error)
                }
            case 304:
                debugPrint(response)
            default:
                single(.error(IssueResponseStatusCodeError(code: statusCode ?? 0)))
            }
        }
    return Disposables.create()
    }

Answer 1

首先你需要改变你的想法。 你没有在应用程序中做任何事情。充其量，您布置 Observable 链（它们不再做任何事情，而水管“做”某事。）然后启动应用程序并让“水”流动。

因此，考虑到这一点，让我们检查您的问题：

我想从服务器获取数据...

并不是“你”想要获取数据。该请求是由于用户的某些操作（可能是按钮点击）或某些其他副作用而产生的。那是什么动作？这需要在代码中表达出来。对于以下内容，我将假设它是一个按钮点击。这意味着您应该：

class Example: UIViewController {
    var button: UIButton!
    var restRepo: RestRepo!

    override func viewDidLoad() {
        super.viewDidLoad()
        let serverResponse = button.rx.tap
            .flatMapLatest { [restRepo] in
                restRepo!.getUnread()
                    .map { Result<[Model], Error>.success($0) }
                    .catchError { .just(Result<[Model], Error>.failure($0)) }
            }
            .share(replay: 1)
    }
}

protocol RestRepo {
    func getUnread() -> Observable<[Model]>
}

struct ProductionRestRepo: RestRepo {
    func getUnread() -> Observable<[Model]> {
        let urlComponent = ApiHelper.instance.dolphinURLComponents(for: ApiHelper.ISSUES_PATH)
        let urlRequest = URLRequest(url: urlComponent.url!)
        return URLSession.shared.rx.data(request: urlRequest)
            .map { try JSONDecoder().decode([Model].self, from: $0) }
    }
}

class ApiHelper {
    static let ISSUES_PATH = ""
    static let instance = ApiHelper()

    func dolphinURLComponents(for: String) -> URLComponents { fatalError() }
}

struct Model: Decodable { }

这里要注意的是getUnread() 是由button.rx.tap 引起的效果。以上建立了因果链。

你的问题继续说“你”想要：

...更新我的数据库...

这里，原因是网络请求，结果是DB保存，所以我们只需将它添加到viewDidLoad（注意下面的代码使用RxEnumKit。）：

let dbResponse = serverResponse
    .capture(case: Result.success)
    .flatMapLatest { [dbRepo] models in
        dbRepo!.save(models)
            .map { Result<Void, Error>.success(()) }
            .catchError { .just(Result<Void, Error>.failure($0)) }
    }

您的问题还表明“您”想要：

...向用户显示接收到的数据。

请注意，向用户显示接收到的数据与数据库保存无关。它们是两个独立的操作，可以并行完成。

将接收到的数据显示给用户，以serverResponse为因，showing为果。

serverResponse
    .capture(case: Result.success)
    .subscribe(onNext: { models in
        print("display the data to the user.", models)
    })
    .disposed(by: disposeBag)

最后，你没有提到它，但你也必须处理错误：

所以也将它添加到 viewDidLoad 中：

Observable.merge(serverResponse.capture(case: Result.failure), dbResponse.capture(case: Result.failure))
    .subscribe(onNext: { error in
        print("an error occured:", error)
    })
    .disposed(by: disposeBag)

---

下面的代码是上面所有的一个块。这编译得很好......

import UIKit
import RxSwift
import RxCocoa
import EnumKit
import RxEnumKit

extension Result: CaseAccessible { }

class Example: UIViewController {
    var button: UIButton!
    var restRepo: RestRepo!
    var dbRepo: DBRepo!
    let disposeBag = DisposeBag()

    override func viewDidLoad() {
        super.viewDidLoad()
        let serverResponse = button.rx.tap
            .flatMapLatest { [restRepo] in
                restRepo!.getUnread()
                    .map { Result<[Model], Error>.success($0) }
                    .catchError { .just(Result<[Model], Error>.failure($0)) }
            }
            .share(replay: 1)

        let dbResponse = serverResponse
            .capture(case: Result.success)
            .flatMapLatest { [dbRepo] models in
                dbRepo!.save(models)
                    .map { Result<Void, Error>.success(()) }
                    .catchError { .just(Result<Void, Error>.failure($0)) }
            }

        serverResponse
            .capture(case: Result.success)
            .subscribe(onNext: { models in
                print("display the data to the user.", models)
            })
            .disposed(by: disposeBag)

        Observable.merge(serverResponse.capture(case: Result.failure), dbResponse.capture(case: Result.failure))
            .subscribe(onNext: { error in
                print("an error occured:", error)
            })
            .disposed(by: disposeBag)
    }
}

protocol RestRepo {
    func getUnread() -> Observable<[Model]>
}

protocol DBRepo {
    func save(_ models: [Model]) -> Observable<Void>
}

struct ProductionRestRepo: RestRepo {
    func getUnread() -> Observable<[Model]> {
        let urlComponent = ApiHelper.instance.dolphinURLComponents(for: ApiHelper.ISSUES_PATH)
        let urlRequest = URLRequest(url: urlComponent.url!)
        return URLSession.shared.rx.data(request: urlRequest)
            .map { try JSONDecoder().decode([Model].self, from: $0) }
    }
}

class ApiHelper {
    static let ISSUES_PATH = ""
    static let instance = ApiHelper()

    func dolphinURLComponents(for: String) -> URLComponents { fatalError() }
}

struct Model: Decodable { }

我希望所有这些对您有所帮助，或者至少会产生更多问题。