今天，笔者与大家分析一下，RxSwift中另一个特别重要，甚至是最重要之一的操作符：flatMap。
如果大家没有读过本系列的第一篇，建议先阅读一下再来看本篇内容。
在RxSwift进行开发的过程中，有一个非常常见的场景:点击某个按钮，然后发送请求，之后处理请求结果。代码大致如下：
myBtn
.rx
.tap
.flatMap { _ in
performRequest()
}
.map {
processResponse($0)
}
.subScribe(
onNext: { ... }
)
.disposed(by: bag)
在这个场景中，所要使用的最核心的一个操作符便是flatMap，它的作用是将myBtn产生的点击事件转换成另一个事件序列，并将该事件序列产生的事件向下传递。代码如下:
extension ObservableType {
public func flatMap<Source: ObservableConvertibleType>(_ selector: @escaping (Element) throws -> Source)
-> Observable<Source.Element> {
return FlatMap(source: self.asObservable(), selector: selector)
}
}
该方法接收一个(Element) throws -> Source的closure，Element就是自身的元素类型，而返回值类型Source是一个泛型参数，并且是遵循ObservableConvertibleType的。也就是说，这个closuer会把上游产生的事件转换成一个事件序列。方法实现中返回了一个FlatMap类型的对象，这个对象持有了self.asObservable()，以及传入的closure.
接下来，咱们看一下这个FlatMap的源码：
final private class FlatMap<SourceElement, SourceSequence: ObservableConvertibleType>: Producer<SourceSequence.Element> {
typealias Selector = (SourceElement) throws -> SourceSequence
private let source: Observable<SourceElement>
private let selector: Selector
init(source: Observable<SourceElement>, selector: @escaping Selector) {
self.source = source
self.selector = selector
}
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == SourceSequence.Element {
let sink = FlatMapSink(selector: self.selector, observer: observer, cancel: cancel)
let subscription = sink.run(self.source)
return (sink: sink, subscription: subscription)
}
}
FlatMap是一个Producer的子类，在它的run方法中通过self.selector和observer生成了一个FlatMapSink类的对象。这个selector就是在调用flatMap时传入的closure, observer其实就是在subscribe方法中创建的observer（参见系列第一篇）。然后调用了FlatMapSink的run方法。但是我们再FlatMapSink里并没有看到run方法，因为FlatMapSink是MergeSink的子类，它的很多逻辑都在它的父类MergeSink里面：
private class MergeSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType>
: Sink<Observer>
, ObserverType where Observer.Element == SourceSequence.Element {
typealias ResultType = Observer.Element
typealias Element = SourceElement
let lock = RecursiveLock()
var subscribeNext: Bool {
true
}
// state
let group = CompositeDisposable()
let sourceSubscription = SingleAssignmentDisposable()
var activeCount = 0
var stopped = false
......
func run(_ source: Observable<SourceElement>) -> Disposable {
_ = self.group.insert(self.sourceSubscription)
let subscription = source.subscribe(self)
self.sourceSubscription.setDisposable(subscription)
return self.group
}
}
在源码的最后面，我们看到了run方法，只有四行，我们需要关注的是第二行：
let subscription = source.subscribe(self)
也就是说让self订阅了传入的Observable，而这个传入的Observable就是FlatMap.source，也就是上游序列。这样一来大致流程就清楚了：
通过Obsevable.flatMap，让FlatMap类持有了上游Obsevable，并通过FlatMapSink订阅了这个上游Obsevable。接下来上游Obsevable的所有事件都会被传递给FlatMapSink的on方法（原理参见系列第一篇）。
接下来的重点，我们需要看一下这个FlatMapSink的on方法做了什么，还是在父类MergeSink里面。
func on(_ event: Event<SourceElement>) {
switch event {
case .next(let element):
if let value = self.nextElementArrived(element: element) {
self.subscribeInner(value.asObservable())
}
case .error(let error):
self.lock.performLocked {
self.forwardOn(.error(error))
self.dispose()
}
case .completed:
self.lock.performLocked {
self.stopped = true
self.sourceSubscription.dispose()
self.checkCompleted()
}
}
}
error事件会直接调用forwardOn，将事件传递给下游observer，completed事件则是调用checkCompleted方法，在checkCompleted方法中判断是否需要将completed事件传递下去。这是因为MergeSink可能会订阅多个事件序列，对这些事件进行一系列变换，直到所有的事件序列都complete之后才会传递，但这个不是今天的重点，所以只要知道个大概就可以了，具体的咱们在后续的文章中再介绍。
接下来看next事件，在next事件中会调用self.nextElementArrived(element: element)：
final private func nextElementArrived(element: SourceElement) -> SourceSequence? {
self.lock.performLocked {
if !self.subscribeNext {
return nil
}
do {
let value = try self.performMap(element)
self.activeCount += 1
return value
}
catch let e {
self.forwardOn(.error(e))
self.dispose()
return nil
}
}
}
这个方法会首先加锁，然后执行self.performMap(element)方法,这个方法在子类(FlatMapSink）中有实现，其实就是调用self.selector：
override func performMap(_ element: SourceElement) throws -> SourceSequence {
try self.selector(element)
}
而这个selector其实就是在最初调用Observable.flatMap时传入的closure，所以到这里，myBtn的点击事件就会被转换成了一个事件序列。获取这个返回值之后，on方法会调用subscribeInner方法：
func subscribeInner(_ source: Observable<Observer.Element>) {
let iterDisposable = SingleAssignmentDisposable()
if let disposeKey = self.group.insert(iterDisposable) {
let iter = MergeSinkIter(parent: self, disposeKey: disposeKey)
let subscription = source.subscribe(iter)
iterDisposable.setDisposable(subscription)
}
}
在这里，主要是创建了一个MergeSinkIter对象，并让这个MergeSinkIter对象订阅了传入的Observable。根据MergeSinkIter的on方法：
func on(_ event: Event<Element>) {
self.parent.lock.performLocked {
switch event {
case .next(let value):
self.parent.forwardOn(.next(value))
case .error(let error):
self.parent.forwardOn(.error(error))
self.parent.dispose()
case .completed:
self.parent.group.remove(for: self.disposeKey)
self.parent.activeCount -= 1
self.parent.checkCompleted()
}
}
}
我们可以看见它实际就是将事件传递给了parent，也就是MergeSink。但是注意，这里是直接调用的forwardOn方法，所以会直接把事件传递给下游observer，而不会触发performMap方法（当然也没法直接调用on方法，因为二者接收的参数泛型通常是不一样的）。
这样一来，flatMap的完整流程就结束了。
总结一下：
1.通过flatMap创建了一个FlatMap对象，并通过FlatMapSink订阅了上游Observable
2.当上游Observable产生一个事件时FlatMapSink通过调用FlatMap.performMap方法触发传入的closure，产生了新的事件序列。
3.通过MergeSinkIter将新的事件序列的事件传递个下游的observer，从而实现了Event->Observable->Event的传递。
它的流程和RxSwift中大部分操作符的核心逻辑差不多，都是通过持有上游observable，创建一个Sink的子类去订阅这个上游observable，然后将事件传递个下游的observer。只不过因为这里产生的是事件序列，所以又多出了一个MergeSinkIter来处理这个事件序列的事件并进行传递。
当然，今天只是介绍了FlatMap的具体流程，MergeSink的相关细节我们都忽略了，这个我们后面再说。
码字不易，若有错漏，欢迎指正。若有助益，烦请点赞。^ _ ^
网友评论