RxSwift-Observable初识

Observable三步曲

使用

//1.注册序列
let ob = Observable<Any>.create { (obserable) -> Disposable in
    //3.发送信号
    obserable.onNext("RXSwift初识")
    obserable.onError(NSError.init(domain: "error", code: 404, userInfo: nil))
    return Disposables.create()
}

//2.订阅信号
let _ = ob.subscribe(onNext: { (value) in
    print("subcribe-\(value)")
}, onError: { (error) in
    print("error-\(error)")
}, onCompleted: {
    print("completed")
}) {
    print("dispos")
}

源码解析

1. 创建序列

//Create.swift
extension ObservableType {
    
    public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
        return AnonymousObservable(subscribe)
    }
}

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

    let _subscribeHandler: SubscribeHandler

    init(_ subscribeHandler: @escaping SubscribeHandler) {
        //核心代码保存创建序列时，传入的闭包subscribe
        self._subscribeHandler = subscribeHandler
    }
    ...
}

Observable的create创建了AnonymousObservable对象，对象内部会保存传入的Handler，AnonymousObservable这个类又继承自Producer，Producer继承了ObservableType协议，并实现了协议的subscribe方法，ObservableType这个协议就一个方法subscribe

2. 订阅信号

//ObservableType+Extensions.swift
public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
            let disposable: Disposable        
            ...   
            //创建AnonymousObserver，保存onNext、onError、onCompleted        
            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(
                //以observer（AnonymousObserver的实例）为参数，调用subscribe，也就是注册时Producer实现的subscribe方法
                self.asObservable().subscribe(observer),
                disposable
            )
    }

final class AnonymousObserver<Element>: ObserverBase<Element> {
    typealias EventHandler = (Event<Element>) -> Void
    
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        //核心代码保存事件
        self._eventHandler = eventHandler
    }
    override func onCore(_ event: Event<Element>) {
        return self._eventHandler(event)
    }
}
//ObserverBase.swift
class ObserverBase<ElementType> : Disposable, ObserverType {
    ...
    func on(_ event: Event<E>) {
        switch event {
        case .next:
            if load(self._isStopped) == 0 {
                self.onCore(event)
            }
        case .error, .completed:
            if fetchOr(self._isStopped, 1) == 0 {
                self.onCore(event)
            }
        }
    }

    func onCore(_ event: Event<E>) {
        rxAbstractMethod()
    }
    ...
}
//Producer.swift
class Producer<Element> : Observable<Element> {
    override init() {
        super.init()
    }
    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()
            //调用AnonymousObservable类的run方法，observer是AnonymousObserver对象
            let sinkAndSubscription = self.run(observer, cancel: disposer)
            disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

            return disposer
        }
        else {
            ...
       }
    }

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

//Create.swift
final private class AnonymousObservable<Element>: Producer<Element> {
    ...
    override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
        let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
        let subscription = sink.run(self)
        return (sink: sink, subscription: subscription)
    }
}

//Create.swift
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
    ...
    func on(_ event: Event<E>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        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()
            }
        }
    }

    func run(_ parent: Parent) -> Disposable {
        return parent._subscribeHandler(AnyObserver(self))
    }
}
//AnyObserver.swift
public struct AnyObserver<Element> : ObserverType {
    /// The type of elements in sequence that observer can observe.
    public typealias E = Element
    
    /// Anonymous event handler type.
    public typealias EventHandler = (Event<Element>) -> Void

    private let observer: EventHandler

    /// Construct an instance whose `on(event)` calls `eventHandler(event)`
    ///
    /// - parameter eventHandler: Event handler that observes sequences events.
    public init(eventHandler: @escaping EventHandler) {
        self.observer = eventHandler
    }
    
    /// Construct an instance whose `on(event)` calls `observer.on(event)`
    ///
    /// - parameter observer: Observer that receives sequence events.
    public init<O : ObserverType>(_ observer: O) where O.E == Element {
        self.observer = observer.on
    }
    
    /// Send `event` to this observer.
    ///
    /// - parameter event: Event instance.
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

    /// Erases type of observer and returns canonical observer.
    ///
    /// - returns: type erased observer.
    public func asObserver() -> AnyObserver<E> {
        return self
    }
}

Observable 调用 subscribe(onNext: , onError:, onCompleted:)，这个方法里初始化了继承自AnonymousObserver的对象observe，AnonymousObserver实现了onCore，onCore会调用_eventHandler, AnonymousObserver的父类ObserverBase实现了on方法，on方法中又会执行onCore，方法里最后会调用 AnonymousObservable的父类Producer实现的方法subscribe，并传入参数是observer；subscribe方法调用AnonymousObservable实现的方法run，传入参数是代入的observer，这个方法里会初始化AnonymousObservableSink对象sink，AnonymousObservableSink实现了on方法，及sink的run方法，下面会调用sink的run，内部又会调用AnonymousObservable的_subscribeHandler，并传入参数AnyObserver(AnonymousObservableSink)，调用AnyObserver的on方法就会调用observer的on方法

3. 发送信号

//ObserverType.swift
extension ObserverType {
    
    /// Convenience method equivalent to `on(.next(element: Element))`
    ///
    /// - parameter element: Next element to send to observer(s)
    public func onNext(_ element: Element) {
        self.on(.next(element))
    }
    
    /// Convenience method equivalent to `on(.completed)`
    public func onCompleted() {
        self.on(.completed)
    }
    
    /// Convenience method equivalent to `on(.error(Swift.Error))`
    /// - parameter error: Swift.Error to send to observer(s)
    public func onError(_ error: Swift.Error) {
        self.on(.error(error))
    }
}

发送信号时会调用on方法，即会调用到_eventHandler，最终处理event！

总结：创建序列时，保存subscribeHandler，在订阅时创建一个AnonymousObserver对象，保存eventHandler，然后再调用之前保存的subscribeHandler，并将AnonymousObserver对象传入，然后在subscribeHandler中执行onNext、onError、onCompleted 等事件时，再调用AnonymousObserver对象中订阅时存储的闭包

思维导图

Obserable

生活如此美好，今天就点到为止。。。