RXSwift - Sink链实现逻辑

这篇文章主要还是要通过 map{} 和 mapsink 来简单剖析 sink链的实现逻辑，来初步感受 RxSwift通过这种机制把Observable的创建和订阅时Observable的行为进行了拆分。

        _ = Observable.just("1").map{ return $0 + "zp"  }.subscribe(onNext: { (str) in
            print(str)
        })

先从 Observable.just("1") 开始

如果想系统了解 just 的调用流程 请点击这里。

  • 通过的 Just 类里面的 extension ObservableType 扩展方法 just , 我们进入了 Just 类中，进行初始化操作。

  • Just 类 init, 保存 just 参数。 self._element = element。

 init(element: Element) {
        self._element = element
    }
  • 接着会调用 Producer Observable 的初始化方法。
// Producer
    override init() {
        super.init()
    }

// Observable
    init() {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
    }

紧接着进入正题 map{} , 这里是重点。

operator 的初始化套路，结合 just 和 map 的初始化我们先小小的总结一下。

首先 - 我们直接调用的 just() 和 map{} 并不是 Just 和 Map 类的初始化方法，而是和它们在一个类的 extension ObservableType 扩展方法。
其次 - Just 和 Map 的初始化方法把相关的参数都会保存成属性，Just 的 _element 以及 Map 的 _source _transform。
最后 - 就是把 Producer 和 Observable 初始化一遍。

  • 通过 extension Observable 的 map<Result>(_ transform: 方法 调用 Map 的 init 方法。
    public func map<Result>(_ transform: @escaping (Element) throws -> Result)
        -> Observable<Result> {
        return Map(source: self.asObservable(), transform: transform)
    }

一、我们先来初始化

  • 通过 init 方法把 _source 和 _transform 保存起来。当然 Producer 和 Observable 初始化一遍。
  • _source 是 map Operator 要处理的 observable, 就是咱们的 just("1")。
  • _transform 就是我们要进行操作的逻辑，也就是 { return $0 + "zp" } 这部分代码。
    typealias Transform = (SourceType) throws -> ResultType

    private let _source: Observable<SourceType>

    private let _transform: Transform

    init(source: Observable<SourceType>, transform: @escaping Transform) {
        self._source = source
        self._transform = transform
    }

二、 我们进入 subscribe(onNext: ... 这个承上启下的方法。

  • 首先是 AnonymousObserver 初始化，并保存 _eventHandler 闭包。
// subscribe  AnonymousObserver  调用

            let observer = AnonymousObserver<Element> { event in
                
                #if DEBUG
                    synchronizationTracker.register(synchronizationErrorMessage: .default)
                    defer { synchronizationTracker.unregister() }
                #endif
                
                switch event {
                case .next(let value):
                    onNext?(value)
                case .error(let error):
                    if let onError = onError {
                        onError(error)
                    }
                    else {
                        Hooks.defaultErrorHandler(callStack, error)
                    }
                    disposable.dispose()
                case .completed:
                    onCompleted?()
                    disposable.dispose()
                }
            }

// AnonymousObserver 初始化
    typealias EventHandler = (Event<Element>) -> Void
    
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        self._eventHandler = eventHandler
    }
  • 其次 return Disposables.create( self.asObservable().subscribe(observer), disposable ) 方法
  • self.asObservable() 是 map, subscribe 直接进入的是 Producer 的方法，因为 Map 没重写直接走的父类。
    override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        if !CurrentThreadScheduler.isScheduleRequired {
            // The returned disposable needs to release all references once it was disposed.
            let disposer = SinkDisposer()
            let sinkAndSubscription = self.run(observer, cancel: disposer)
            disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
                disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

                return disposer
            }
        }
    }

三、sink 调用入口就在这。

  • 先不管scheduler 相关的判断, 直接先看 SinkDisposer 和 run 方法。
    let disposer = SinkDisposer()
    let sinkAndSubscription = self.run(observer, cancel: disposer)
    disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
  • 为了方便理解 ，我们先看 run 方法。因为 Map 重写了 run 方法所以必然进入 Map 的 run。而 run 就是 sink 链的 入口

问：run 两个入参 Observer 和 disposer 都是谁
答：disposer 毋庸置疑就是 SinkDisposer 对象， Observer 是 AnonymousObserver<String>。

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == ResultType {
        let sink = MapSink(transform: self._transform, observer: observer, cancel: cancel)
        let subscription = self._source.subscribe(sink)
        return (sink: sink, subscription: subscription)
    }
  • MapSink 初始化是把 _transform 存在自己的属性中， observer、 cancel 是保存在父类 Sink 中的。
// MapSink
    typealias Transform = (SourceType) throws -> ResultType

    typealias ResultType = Observer.Element 
    typealias Element = SourceType

    private let _transform: Transform

    init(transform: @escaping Transform, observer: Observer, cancel: Cancelable) {
        self._transform = transform
        super.init(observer: observer, cancel: cancel)
    }

// Sink
    fileprivate let _observer: Observer
    fileprivate let _cancel: Cancelable
    private let _disposed = AtomicInt(0)

    #if DEBUG
        private let _synchronizationTracker = SynchronizationTracker()
    #endif

    init(observer: Observer, cancel: Cancelable) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        self._observer = observer
        self._cancel = cancel
    }

四、睁大你布灵布灵的大眼睛，花了呼哨的操作要开始了

  • 随着 MapSink 初始化结束， 我们走到了 let subscription = self._source.subscribe(sink)这个方法

问：self._source是谁？subscribe 是谁的方法？
答：self._source 是初始化时就存下的 just("1"), 因为 Just override 了 subscribe,所以 subscribe 就是 Just 的无疑了。

  • 我们进入 Just 的 subscribe 方法，并执行第一个方法 observer.on(.next(self._element))

问：observer 是谁？，self._element 是谁？. next 又是什么玩意？ on 调用谁的？
答：
observer是在 Map 调用 run 方法是传进来的 MapSink 对象。
self._element 就是在 Just 初始化时存下的 “1”。
.next 就是 enum Event<Element> 。
on 方法自然就是 MapSink 的方法喽。

    override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        observer.on(.next(self._element)) // self._element 也就是 “1”
        observer.on(.completed)
        return Disposables.create()
    }
  • 下面我们进入 MapSink 的 on 方法，进入 case .next(let element) 方法, self._transform 就是 Map 执行的逻辑部分，element 是在 Just 类的 subscribe 的 on 穿过来的 “1”。
    func on(_ event: Event<SourceType>) {
        switch event {
        case .next(let element):
            do {
                let mappedElement = try self._transform(element)
                self.forwardOn(.next(mappedElement))
            }
            catch let e {
                self.forwardOn(.error(e))
                self.dispose()
            }
        case .error(let error):
            self.forwardOn(.error(error))
            self.dispose()
        case .completed:
            self.forwardOn(.completed)
            self.dispose()
        }
    }
  • 进入 Sink 的 forwardOn 。 forwardOn 是不能重写的，所以直接进入父类的。这里 self._observer.on(event) 的 self._observer 是 AnonymousObserver , 这个 on 方法则是 AnonymousObserver 的父类 ObserverBase 。

这里需要捋一下 self._observer 和 on 是谁？ 怎么来的？

  • self._observer
    首先我们要追溯到很久的 extension ObservableType 的 subscribe(onNext 方法我们调用了 self.asObservable().subscribe(observer) ， 这个方法可以理解成 Map().subscribe(AnonymousObserver()) 就是调用map 的 subscribe 方法但是 map 没有重写，直接进入了它的父类 Producer 的 subscribe ,在这个方法里面调用了 self.run(observer, cancel: disposer) 方法，它可以理解成 map.run(AnonymousObserver(),SinkDisposer()),就是 Map 的 run方法，在这个方法我们初始化了 MapSink 而入参 Observable 就是 AnonymousObserver对象。
  • self._observer简图
    subscribe(onNext(extension ObservableType) -> subscribe (Producer) -> run(map) -> init(MapSink)
  • on 既然 self._observer 是 AnonymousObserver , on 理应是 AnonymousObserver 的，但是它没重写 on, 所以实际 调用的事它的父类 ObserverBase 的 on 方法。
    final func forwardOn(_ event: Event<Observer.Element>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        if isFlagSet(self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }

五、从这里开始就接近尾声了

  • ObserverBase 的 on 方法，调用 self.onCore(event) , 所以进入 AnonymousObserver 的 onCore。onCore 又调用了 self._eventHandler(event) 闭包，又回调到 extension ObservableType 的 subscribe(onNext: ((Element) ... 的 57行，然后就是通过 onNext onError onCompleted onDisposed 回到到最初的地方。
// ObserverBase

    func on(_ event: Event<Element>) {
        switch event {
        case .next:
            if load(self._isStopped) == 0 {
                self.onCore(event)
            }
        case .error, .completed:
            if fetchOr(self._isStopped, 1) == 0 {
                self.onCore(event)
            }
        }
    }

// AnonymousObserver

    override func onCore(_ event: Event<Element>) {
        return self._eventHandler(event)
    }

// extension ObservableType

 public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
            let disposable: Disposable
            
            if let disposed = onDisposed {
                disposable = Disposables.create(with: disposed)
            }
            else {
                disposable = Disposables.create()
            }
            
            #if DEBUG
                let synchronizationTracker = SynchronizationTracker()
            #endif
            
            let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
            
            let observer = AnonymousObserver<Element> { event in
                
                #if DEBUG
                    synchronizationTracker.register(synchronizationErrorMessage: .default)
                    defer { synchronizationTracker.unregister() }
                #endif
                
                switch event {
                case .next(let value):
                    onNext?(value)
                case .error(let error):
                    if let onError = onError {
                        onError(error)
                    }
                    else {
                        Hooks.defaultErrorHandler(callStack, error)
                    }
                    disposable.dispose()
                case .completed:
                    onCompleted?()
                    disposable.dispose()
                }
            }
            return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )
    }

六、 SinkDisposer 是干嘛的？

我们可以把 disposer 就是回收资源了， run 笼统理解成运行。

    let disposer = SinkDisposer()
    let sinkAndSubscription = self.run(observer, cancel: disposer)
    disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
  • 在 SinkDisposer 的 setSinkAndSubscription 中我们保存了， sinkAndSubscription.sink 和 sinkAndSubscription.subscription。
  • sinkAndSubscription.sink 就是 MapSink
  • sinkAndSubscription.subscription 就是 Just 的 Disposables.create()

SinkDisposer可以一边调用_sink.dispose()释放掉Sink派生类占据的资源，另一边调用_subscription.dispose()释放掉订阅上游事件时占据的资源；完成“双边释放”之后，它会把这两个optionals设置为nil

  • dispose 方法调用的时候 ，会把它持有的相关资源释放，并把自己置为 nil 
  func dispose() {
        let previousState = fetchOr(self._state, DisposeState.disposed.rawValue)

        if (previousState & DisposeState.disposed.rawValue) != 0 {
            return
        }

        if (previousState & DisposeState.sinkAndSubscriptionSet.rawValue) != 0 {
            guard let sink = self._sink else {
                rxFatalError("Sink not set")
            }
            guard let subscription = self._subscription else {
                rxFatalError("Subscription not set")
            }

            sink.dispose()
            subscription.dispose()

            self._sink = nil
            self._subscription = nil
        }
    }
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 216,843评论 6赞 502
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 92,538评论 3赞 392
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 163,187评论 0赞 353
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,264评论 1赞 292
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,289评论 6赞 390
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,231评论 1赞 299
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,116评论 3赞 418
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,945评论 0赞 275
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,367评论 1赞 313
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,581评论 2赞 333
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,754评论 1赞 348
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,458评论 5赞 344
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,068评论 3赞 327
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,692评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,842评论 1赞 269
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,797评论 2赞 369
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,654评论 2赞 354