RxSwift：share() 替代方案，保证upstream 的单一订阅

发布于05月24日

我一直认为，无论下游用户有多少，.share(replay: 1, scope: .forever)股都是单一的upstream 认购.

然而，我刚刚发现，如果下游订阅数降到零，它将停止"共享"，并在upstream 释放订阅(因为在引擎盖下使用refCount()).因此，当一个新的下游订阅发生时，它必须在upstream re-subscribe.在以下示例中:

let sut = Observable<Int>
    .create { promise in
        print("create")
        promise.onNext(0)
        return Disposables.create()
    }
    .share(replay: 1, scope: .forever)

sut.subscribe().dispose()
sut.subscribe().dispose()

我希望create只打印一次，但它会打印两次.如果我删除.dispose()个电话——就一次.

如何建立保证upstream 最多订阅一次的链？

推荐答案

您描述的目标意味着您应该使用multicast(或使用它的操作符之一，如publish()、replay(_:)或replayAll())，而不是share...

let sut = Observable<Int>
    .create { observer in
        print("create")
        observer.onNext(0)
        return Disposables.create()
    }
    .replay(1)

let disposable = sut.connect() // subscription will stay alive until dispose() is called on this disposable...

sut.debug("one").subscribe().dispose()
sut.debug("two").subscribe().dispose()

了解两者之间的区别.永远和.连接时，请阅读"ShareReplayScope.swift"文件中的文档.两者都是重新计数的，但区别在于如何处理重新订阅操作符.下面是一些测试代码来显示差异...

class SandboxTests: XCTestCase {
    var scheduler: TestScheduler!
    var observable: Observable<String>!

    override func setUp() {
        super.setUp()
        scheduler = TestScheduler(initialClock: 0)
        // creates an observable that will error on the first subscription, then call `.onNext("A")` on the second.
        observable = scheduler.createObservable(timeline: "-#-A")
    }

    func testWhileConnected() {
        // this shows that re-subscription gets through the while connected share to the source observable
        let result = scheduler.start { [observable] in
            observable!
                .share(scope: .whileConnected)
                .retry(2)
        }
        XCTAssertEqual(result.events, [
            .next(202, "A")
        ])
    }

    func testForever() {
        // however re-subscription doesn't get through on a forever share
        let result = scheduler.start { [observable] in
            observable!
                .share(scope: .forever)
                .retry(2)
        }
        XCTAssertEqual(result.events, [
            .error(201, NSError(domain: "Test Domain", code: -1))
        ])
    }
}