RxSwift(2) - 核心流程分析

2019-08-19 本文已影响0人 H_Zimi

RxSwift 事件 Event

next— 序列产生了一个新的元素

error — 创建序列时产生了一个错误，导致序列终止

completed — 序列的所有元素都已经成功产生，整个序列已经完成

序列的创建

//1、创建序列
let ob = Observable<Any>.create { (anyObserver) -> Disposable in
    //3、发送信号
    anyObserver.onNext("RxSwift")
    anyObserver.onError(NSError(domain: "com.zm", code: 10087, userInfo: nil))
    anyObserver.onCompleted()
    return Disposables.create()
}

public class Observable<Element> : ObservableType{
  //省略类的具体定义，我们主要看类的继承关系和遵守的协议
  //类中的具体内容，可以在探索的过程中跳过去查看
}

Observable遵守协议ObservableType；
Observable没有实现create，而是调用的是ObservableType协议可选方法create；
subscribe闭包实现，闭包参数为AnyObserver类型的对象anyObserver，这里没有给这个对象命名为observer是为了与后面流程中以示区别。

 public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
        return AnonymousObservable(subscribe)
 }

subscribe闭包传值参数类型AnyObserver<Element>；
返回私有的AnonymousObservable匿名可观察序列，这时ob 实际是一个AnonymousObservable类型的序列了；
将subscribe闭包保存在AnonymousObservable的属性_subscribeHandler中，完成序列的创建。
AnonymousObservable继承Producer.

final private class AnonymousObservable<Element>: Producer<Element> {
    typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable

    let _subscribeHandler: SubscribeHandler

    init(_ subscribeHandler: @escaping SubscribeHandler) {
        self._subscribeHandler = subscribeHandler
    }

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
        let subscription = sink.run(self)
        return (sink: sink, subscription: subscription)
    }
}

发送订阅

//2、发送订阅
let _ = ob.subscribe(onNext: { (text) in
     print("\(text)")
}, onError: { (error) in
     print("\(error)")
 }, onCompleted: {
     print("完成")
 }) {
     print("销毁")
}

subscribe发送订阅，注意这里的subscribe是方法，创建序列的时候的subscribe是闭包；

//发送订阅的方法
 public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
      //省略了销毁者相关代码，可以探索时在源码中查看
      let observer = AnonymousObserver<Element> { event in
          //省略部分无关代码   
          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
       )
 }

创建了一个AnonymousObserver匿名观察者对象observer；
AnonymousObserver继承ObserverBase，ObserverBase遵守ObserverType协议；
asOservable()确保是一个序列；
subscribe在AnonymousObservable中没有实现，调用的是父类Producer中的subscribe方法，并将observer传入其中。

override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
    if !CurrentThreadScheduler.isScheduleRequired {
     //省略部分与分析流程无关代码
    }
    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
         }
     }
}

demo不涉及开线程情况，走当前线程
调用AnonymousObservable的run方法，传入observer

//AnonymousObservable中的run方法
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
    let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
    //调用AnonymousObservableSink中run方法
    let subscription = sink.run(self)
    return (sink: sink, subscription: subscription)
    }

此时又引入了一个新的类AnonymousObservableSink，起到关联观察者Observer与可观察序列Observable的作用；
AnonymousObservableSink继承Sink类，并遵守了ObserverType协议；
创建AnonymousObservableSink的对象sink，在初始化时又将run方法传递过来的observer，cancel交由父类Sink保存在_observer、_cancel中;
调用sink的run方法将AnonymousObservable自己传入；

typealias Parent = AnonymousObservable<Element>
//AnonymousObservableSink中的run方法
func run(_ parent: Parent) -> Disposable {
   return parent._subscribeHandler(AnyObserver(self)
 }

从源码中可以看出Parent是AnonymousObservable的别名，parent是一个AnonymousObservable对象；
在run方法的返回值中，执行了parent的_subscribeHandler闭包回调，并将AnonymousObservableSink对象本身作为AnyObserver初始化方法的参数；

    public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        self.observer = observer.on
    }

在AnyObserver的初始化方法中，将AnonymousObservableSink的on方法保存在了属性observer，注意这里有两个observer，它们不是一样的，跟前文的observer也不一样；

发送消息

//截取Demo中消息发送片段的代码
 observer.onNext("RxSwift")
 observer.onError(NSError(domain: "com.zm", code: 10087, userInfo: nil))
 observer.onCompleted()

func on(_ event: Event<Element>)

public func onNext(_ element: Element) {
    self.on(.next(element))
}
public func onCompleted() {
    self.on(.completed)
    }
public func onError(_ error: Swift.Error) {
    self.on(.error(error))
}

onNext、onCompleted、onError是ObserverType协议协议拓展方法；
ObserverType中定义了on方法，没有具体实现，交有子类AnyObserver实现；

public func on(_ event: Event<Element>) {
    return self.observer(event)
}

执行observer闭包的回调，前文说过self.observe保存的是AnonymousObservableSink中的on方法，那么接下来的流程肯定会走到on方法的具体实现

func on(_ event: Event<Element>) {
//省略部分代码
     switch event {
     case .next:
        if load(self._isStopped) == 1 {
           return
        }
        self.forwardOn(event)
     case .error, .completed:
        if fetchOr(self._isStopped, 1) == 0 {
           self.forwardOn(event)
           self.dispose()
        }
     }
}

on方法中调用了forwardOn方法，forwardOn是在AnonymousObservableSink的父类Sink中实现的

final func forwardOn(_ event: Event<Observer.Element>) {
   //省略部分无关代码
   if isFlagSet(self._disposed, 1) {
       return
    }
    self._observer.on(event)
}

self._observer序列的订阅中已经说过是一个AnonymousObserver对象，on方法在AnonymousObserver没有实现，在它的父类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的onCore方法，执行_eventHandler闭包的回调

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

此时终于回到了subscribe订阅方法中的创建的observer对象是的闭包实现中了：

let observer = AnonymousObserver<Element> { event in
    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()
    }
}

在event事件中执行subscribe方法中闭包参数onNext、onError、onCompleted的回调，完成元素的传递。

总结：

序列的创建过程是将一个观察者对象保存到一个subscribe闭包的过程，在闭包的实现中onNext产生元素，元素完成时onComplete，产生元素时出错的处理onError；

序列的订阅

最后附一张整个流程图

RxSwift(2) - 核心流程分析

序列的创建

发送订阅

发送消息

总结：

猜你喜欢

热点阅读