swift - 为什么完成事件会阻止此测试通过？

Question

我意识到这里有很多关于 SO 问题的代码，但这是我目前能做的最好的...您只需将代码复制/粘贴到支持 rx 的 Playground 中即可查看问题。

第 89 行有一段代码被注释掉了 let creds = Observable.just(credentials)//.concat(Observable.never())。如果我删除 // 并允许 concat，代码将通过测试。如果允许 creds 发送完成事件，任何人都可以提供关于为什么此代码无法通过测试的线索吗？

import Foundation
import RxSwift
import RxCocoa
import UIKit

typealias Credentials = (email: String, password: String)

struct User {
    let id: String
    let properties: [Property]
}

struct Property {
    let id: String
    let name: String
}

struct LoginParams {
    let touchIDPossible: Bool
}

class LoginScreen {
    var attemptLogin: Observable<Credentials> {
        assert(_attemptLogin == nil)
        _attemptLogin = PublishSubject()
        return _attemptLogin!
    }

    var _attemptLogin: PublishSubject<(email: String, password: String)>?
}

class DashboardScreen {
    func display(property: Observable<Property?>) {
        property.subscribe(onNext: { [unowned self] in
            self._property = $0
        }).disposed(by: bag)
    }

    var _property: Property?
    let bag = DisposeBag()
}

class Interface {
    func login(params: LoginParams) -> Observable<LoginScreen> {
        assert(_login == nil)
        _login = PublishSubject()
        return _login!
    }

    func dashboard() -> Observable<DashboardScreen> {
        assert(_dashboard == nil)
        _dashboard = PublishSubject()
        return _dashboard!
    }

    var _login: PublishSubject<LoginScreen>?
    var _dashboard: PublishSubject<DashboardScreen>?
    let bag = DisposeBag()
}

class Server {
    func user(credentials: Credentials) -> Observable<User> {
        assert(_user == nil)
        _user = PublishSubject()
        return _user!
    }

    func property(id: String) -> Observable<Property> {
        assert(_property == nil)
        _property = PublishSubject()
        return _property!
    }

    var _user: PublishSubject<User>?
    var _property: PublishSubject<Property>?
}

class Coordinator {

    init(interface: Interface, server: Server) {
        self.interface = interface
        self.server = server
    }

    func start() {
        let credentials = (email: "foo", password: "bar")

        // remove the `//` and the test will pass. Why does it fail when `creds` completes?
        let creds = Observable.just(credentials)//.concat(Observable.never())

        let autoUser = creds.flatMap {
            self.server.user(credentials: $0)
                .materialize()
                .filter { !$0.isCompleted }
            }.shareReplayLatestWhileConnected()

        let login = autoUser.filter { $0.error != nil }
            .flatMap { _ in self.interface.login(params: LoginParams(touchIDPossible: false)) }

        let attempt = login.flatMap { $0.attemptLogin }
            .shareReplayLatestWhileConnected()

        let user = attempt.flatMap {
            self.server.user(credentials: $0)
                .materialize()
                .filter { !$0.isCompleted }
            }.shareReplayLatestWhileConnected()

        let propertyID = Observable.merge(autoUser, user).map { $0.element }
            .filter { $0 != nil }.map { $0! }
            .map { $0.properties.sorted(by: { $0.name < $1.name }).map({ $0.id }).first }

        let property = propertyID.filter { $0 != nil }.map { $0! }
            .flatMap { self.server.property(id: $0)
                .map { Optional.some($0) }
                .catchErrorJustReturn(nil)
            }.debug("property").shareReplayLatestWhileConnected()

        let dashboard = property.flatMap { _ in self.interface.dashboard() }

        dashboard.map { $0.display(property: property) }
            .subscribe()
            .disposed(by: bag)
    }

    let interface: Interface
    let server: Server
    let bag = DisposeBag()
}

do {
    let interface = Interface()
    let server = Server()
    let coordinator = Coordinator(interface: interface, server: server)

    coordinator.start()

    assert(server._user != nil)

    let simpleProperty = Property(id: "bar", name: "tampa")
    let user = User(id: "foo", properties: [simpleProperty])
    server._user?.onNext(user)
    server._user?.onCompleted()
    server._user = nil

    assert(interface._login == nil)

    assert(server._property != nil)

    let property = Property(id: "bar", name: "tampa")
    server._property!.onNext(property)
    server._property!.onCompleted()
    server._property = nil

    assert(interface._dashboard != nil)

    let dashboard = DashboardScreen()
    interface._dashboard?.onNext(dashboard)
    interface._dashboard?.onCompleted()

    assert(dashboard._property != nil)
    print("test passed")
}

这是上面代码的输出:

2017-06-01 22:22:42.534: property -> subscribed
2017-06-01 22:22:42.552: property -> Event next(Optional(__lldb_expr_134.Property(id: "bar", name: "tampa")))
2017-06-01 22:22:42.557: property -> Event completed
2017-06-01 22:22:42.557: property -> isDisposed
2017-06-01 22:22:42.559: property -> subscribed
assertion failed: file MyPlayground.playground, line 159

为什么 property 在 被处置后被订阅？

如果您删除 \\，这里是输出:

2017-06-01 22:23:51.540: property -> subscribed
2017-06-01 22:23:51.553: property -> Event next(Optional(__lldb_expr_136.Property(id: "bar", name: "tampa")))
test passed

Answer 1

我最初建议将 dashboard 保留在 DisposeBag 中，这样当 start() 完成时，引用不会过早消失。 OP 此后更新了代码，因此这里是对答案的更新尝试。

---

当您添加更多调试信息时:

let dashboard = property.debug("prop in")
    .flatMap { _ in self.interface.dashboard().debug("dash in") }
    .debug("dash out")

日志将显示该属性提前完成，即在订阅了内部序列之后(“dash in -> subscribed”):

2017-06-03 08:33:27.442: property -> Event next(Optional(Property(id: "bar", name: "tampa")))
2017-06-03 08:33:27.442: prop in -> Event next(Optional(Property(id: "bar", name: "tampa")))
2017-06-03 08:33:27.449: dash in -> subscribed
2017-06-03 08:33:27.452: property -> Event completed
2017-06-03 08:33:27.452: property -> isDisposed
2017-06-03 08:33:27.452: prop in -> Event completed
2017-06-03 08:33:27.452: prop in -> isDisposed
2017-06-03 08:33:27.456: dash in -> Event next(DashboardScreen)
2017-06-03 08:33:27.456: dash out -> Event next(DashboardScreen)

如果您使用 .concat(.never()) ，则不会触发完成事件并且不会干扰该过程。

问题是您的测试代码是命令式编写的。您对流程进行 start()，然后发布更改。但是，如果您将各种 onNext 事件异步放入主队列，整个事情就会分崩离析。您的协调器的设计读起来像声明性代码，但实际上就像花哨的命令式顺序代码路径一样使用。

补救措施是考虑及时性。PublishSubjects 没有历史；如果您使用 BehaviorSubjects 来重播它们的最新值，您可以在调用 start() 之前设置所有更改，它会起作用。我假设您使用 PublishSubject s，因为您首先调用 start() 来打开管道并希望通过它一个接一个地推送更改。问题是，你的管道是用一种不等你把所有东西都推过去的方式制造的。输入阀独立关闭，可以这么说。

是的，这个比喻在整个人类历史上都不是最好的:)

所以选项真的是:

1. 将 Coordinator 的所有工作变成一个大的 Observable.combineLatest，以便在每个序列都有发言权之前不会启动整个转换序列，
2. 使用缓冲/重放主题并提前设置，
3. 用永不完成的基本序列替换 .just(完成)以保持管道打开；您可以将其设为 Observable<Observable<Credentials>>，其中外部序列保持事件状态，内部序列使用 Observable.just——尽管我怀疑您的生产代码完全依赖于这个小细节。