本文假设你已经了解了Rxswift的基本概念,且对于Observable,Observer,Operator有一定的了解
来看看下图这个完整的事件流是怎么"串"起来的(暂不考虑dispose)
_ = Observable<Int>.create { observer in
observer.onNext(2)
observer.onCompleted()
return Disposables.create()
}
.map{
$0 * 2
}.filter {
$0 < 1
}.subscribe {
print($0)
}
我们的代码按照如下顺序执行
1.png
先看下面3张图,是RxSwift对ObservableType提供的默认方法,注意红框内:
map方法和filter方法,create方法.png
- 我们调用的map和filter方法内部其实都是直接返回了一个Map或者Filter对象(create也是直接返回了一个对象,但名字有些不一样,为了方便理解,暂时先不管这个)
- 先看第二个参数,它只是保存了你写在map或者filter后面的闭包
- 再来看在Map或者Filter中第一个参数source,指它的上游对象,对于Map来说,就是create返回的结果,即AnonymousObservable对象,不信可以打印出create操作之后的类型,能看到结果如下
let ob = Observable<Int>.create { observer in
observer.onNext(2)
observer.onCompleted()
return Disposables.create()
}
print(type(of: ob))
//打印值:AnonymousObservable<Int>
同理对于Filter来说,它的source就是Map返回的结果了,这应该是个Map类型,下面验证一下
let ob = Observable<Int>.create { observer in
observer.onNext(2)
observer.onCompleted()
return Disposables.create()
}.map{
$0 * 2
}
print(type(of: ob))
//打印值:Map<Int, Int>
其实RxSwift中的操作符都是这个套路.
所以各个内部生成的对象就会形成了对上游的引用链,这时如下图:
接下来先看下面的图
image.png
可以看到Rxswift中的操作符都继承与Producer类,所以我们得先看下Producer类(为了方便理解,这里删减了一些代码)
class Producer<Element>: Observable<Element> {
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
//1
let disposer = SinkDisposer()
//2:run方法留给子类去实现
let sinkAndSubscription = self.run(observer, cancel: disposer)
//3
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
}
其实Producer类也只是重写了Observable的subscribe,而该方法里面的第2步的run方法,是需要由Producer的子类实现的,也就是说:Map,Filter,First等等操作符就是重写run方法,在内部实现自己的核心逻辑的.
这里拿Map举例,其他操作符类似.
final private class Map<SourceType, ResultType>: Producer<ResultType> {
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
}
//重写run方法实现自己的map逻辑
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == ResultType {
//1:创建一个叫做MapSink类的对象,先知道这一点,稍后来说这个MapSink
let sink = MapSink(transform: self.transform, observer: observer, cancel: cancel)
//2 :注意这一步的self.source就是之前引用的它的上游对象
// 这一步就是使用一个叫做sink的对象去订阅它上游事件
//同理它的上游操作符类的run方法里面也会使用这种方法订阅它的上上游事件
//这样它就把"订阅"给串起来了
let subscription = self.source.subscribe(sink)
return (sink: sink, subscription: subscription)
}
}
说完了订阅是如何"串"起来的,下面来说说序列发出的事件是如何层层往下传递的
我们来看看create方法吧
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
AnonymousObservable(subscribe)
}
首先它返回的不是Create对象,而是AnonymousObservable对象
那么再看看AnonymousObservable
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self.subscribeHandler = subscribeHandler
}
//还是老套路,重写了run
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
//同样的它也创建了一个叫做"某某Sink"的类,保存了observer
//这里的observer就是它的下游观察者,还记得上面的MapSink吗,不记得可以回去看看,它里面也有一个observer
//这个observer就是用来保存下游观察者的
//其实每个操作符类的run方法里面都是这种套路:先创建一个"某某Sink"的类,而这个类就是该操作符真正实现逻辑并且传递到下游的地方
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
同样的它也创建了一个叫做"某某Sink"的类,保存了observer,这里的observer就是它的下游观察者,还记得上面的MapSink吗,不记得可以回去看看,它里面也有一个observer
其实每个操作符类的run方法里面都是这种套路:先创建一个"某某Sink"的类,而这个类就是该操作符真正实现逻辑并且传递到下游的地方
我们先来看看AnonymousObservableSink这个Sink类
//删减了一些影响阅读的代码
final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
// 我们在create{}闭包里调用的observer.onNext(2)其实就是走了这个on方法
// _ = Observable<Int>.create { observer in
// observer.onNext(2)
// observer.onCompleted()
// return Disposables.create()
// }
func on(_ event: Event<Element>) {
switch event {
case .next:
if load(self.isStopped) == 1 {
return
}
//forwardOn方法就是向下游Observer发送事件
self.forwardOn(event)
case .error, .completed:
if fetchOr(self.isStopped, 1) == 0 {
//forwardOn方法就是向下游Observer发送事件
self.forwardOn(event)
self.dispose()
}
}
}
func run(_ parent: Parent) -> Disposable {
//subscribeHandler就是create闭包,这里就是执行了这个闭包,并且把AnyObserver(self)作为参数传到闭包
parent.subscribeHandler(AnyObserver(self))
}
}
public struct AnyObserver<Element> : ObserverType {
public typealias EventHandler = (Event<Element>) -> Void
private let observer: EventHandler
public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
//这里的on就是上面AnonymousObservableSink里面的on方法
// 我们在create{}闭包里调用的observer.onNext(2)其实就是走了这个on方法
// _ = Observable<Int>.create { observer in
// observer.onNext(2)
// observer.onCompleted()
// return Disposables.create()
// }
self.observer = observer.on
}
public func on(_ event: Event<Element>) {
self.observer(event)
}
}
on方法内部的forwardOn方法就是向下游Observer发送事件
上面我们说过:每个操作符类的run方法都会创建一个叫做"某某Sink"的类用来处理操作符真正的逻辑
而这些"某某Sink"的类都遵循了ObserverType协议,这就意味着它们都必须实现一个自己的on方法
比如MapSink内的on方法
final private class MapSink<SourceType, Observer: ObserverType>: Sink<Observer>, ObserverType {
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)
}
func on(_ event: Event<SourceType>) {
switch event {
case .next(let element):
do {
//这个transform就是我们写在map闭包里面的转换逻辑
//先执行转换逻辑,然后调用forwardOn把结果转交给下游的Observer
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()
}
}
}
forwardOn定义在了Sink基类内部,它只是调用了下游Observer的on方法而已,如下所示
//为了方便,删除了一些代码
class Sink<Observer: ObserverType>: Disposable {
fileprivate let observer: Observer
fileprivate let cancel: Cancelable
init(observer: Observer, cancel: Cancelable) {
self.observer = observer
self.cancel = cancel
}
final func forwardOn(_ event: Event<Observer.Element>) {
self.observer.on(event)
}
}
