就问此时此刻还有谁?45度仰望天空,该死!我这无处安放的魅力!
- RxSwift(1)—— 初探
- RxSwift(2)—— 核心逻辑源码分析
- RxSwift(3)—— Observable序列的创建方式
- RxSwift(4)—— 高阶函数(上)
- RxSwift(5)—— 高阶函数(下)
- RxSwift(6)—— scheduler源码解析(上)
- RxSwift(7)—— scheduler源码解析(下)
- RxSwift(8)—— KVO底层探索(上)
- RxSwift(9)—— KVO底层探索(下)
- RxSwift(10)—— 场景序列总结
- RxSwift(11)—— 销毁者-dispose源码解析
- RxSwift(12)—— Subject即攻也守
- RxSwift(13)—— 爬过的坑
- RxSwift(14)—— MVVM双向绑定
RxSwift目录直通车--- 和谐学习,不急不躁!
这一篇文章继续上一篇:RxSwift-高阶函数(上)给大家介绍
RxSwift
非常重要的高阶函数,也不多说,开始介绍
5:从可观察对象的错误通知中恢复的操作符
5.1:catchErrorJustReturn
- 从错误事件中恢复,方法是返回一个可观察到的序列,该序列发出单个元素,然后终止
print("*****catchErrorJustReturn*****")
let sequenceThatFails = PublishSubject<String>()
sequenceThatFails
.catchErrorJustReturn("Cooci")
.subscribe { print($0) }
.disposed(by: disposeBag)
sequenceThatFails.onNext("Hank")
sequenceThatFails.onNext("Kody") // 正常序列发送成功的
//发送失败的序列,一旦订阅到位 返回我们之前设定的错误的预案
sequenceThatFails.onError(self.lgError)
5.2:catchError
- 通过切换到提供的恢复可观察序列,从错误事件中恢复
print("*****catchError*****")
let recoverySequence = PublishSubject<String>()
sequenceThatFails
.catchError {
print("Error:", $0)
return recoverySequence // 获取到了错误序列-我们在中间的闭包操作处理完毕,返回给用户需要的序列(showAlert)
}
.subscribe { print($0) }
.disposed(by: disposeBag)
sequenceThatFails.onNext("Hank")
sequenceThatFails.onNext("Kody") // 正常序列发送成功的
sequenceThatFails.onError(lgError) // 发送失败的序列
recoverySequence.onNext("CC")
5.3:retry
- 通过无限地重新订阅可观察序列来恢复重复的错误事件
print("*****retry*****")
var count = 1 // 外界变量控制流程
let sequenceRetryErrors = Observable<String>.create { observer in
observer.onNext("Hank")
observer.onNext("Kody")
observer.onNext("CC")
if count == 1 {
// 流程进来之后就会过度-这里的条件可以作为出口,失败的次数
observer.onError(self.lgError) // 接收到了错误序列,重试序列发生
print("错误序列来了")
count += 1
}
observer.onNext("Lina")
observer.onNext("小雁子")
observer.onNext("婷婷")
observer.onCompleted()
return Disposables.create()
}
sequenceRetryErrors
.retry()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
5.4:retry(_:):
- 通过重新订阅可观察到的序列,重复地从错误事件中恢复,直到重试次数达到
max
未遂计数
print("*****retry(_:)*****")
let sequenceThatErrors = Observable<String>.create { observer in
observer.onNext("Hank")
observer.onNext("Kody")
observer.onNext("CC")
if count < 5 { // 这里设置的错误出口是没有太多意义的额,因为我们设置重试次数
observer.onError(self.lgError)
print("错误序列来了")
count += 1
}
observer.onNext("Lina")
observer.onNext("小雁子")
observer.onNext("婷婷")
observer.onCompleted()
return Disposables.create()
}
sequenceThatErrors
.retry(3)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
6:Rx流程操作符。
6.1:debug
- 打印所有订阅、事件和处理。
print("*****debug*****")
var count = 1
let sequenceThatErrors = Observable<String>.create { observer in
observer.onNext("Hank")
observer.onNext("Kody")
observer.onNext("CC")
if count < 5 {
observer.onError(self.lgError)
print("错误序列来了")
count += 1
}
observer.onNext("Lina")
observer.onNext("小雁子")
observer.onNext("可心")
observer.onCompleted()
return Disposables.create()
}
sequenceThatErrors
.retry(3)
.debug()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
6.2: RxSwift.Resources.total:
- 提供所有Rx资源分配的计数,这对于在开发期间检测泄漏非常有用。
print("*****RxSwift.Resources.total*****")
print(RxSwift.Resources.total)
let subject = BehaviorSubject(value: "Cooci")
let subscription1 = subject.subscribe(onNext: { print($0) })
print(RxSwift.Resources.total)
let subscription2 = subject.subscribe(onNext: { print($0) })
print(RxSwift.Resources.total)
subscription1.dispose()
print(RxSwift.Resources.total)
subscription2.dispose()
print(RxSwift.Resources.total)
7:链接操作符
7.1:multicast
- 将源可观察序列转换为可连接序列,并通过指定的主题广播其发射。
func testMulticastConnectOperators(){
print("*****multicast*****")
let subject = PublishSubject<Any>()
subject.subscribe{print("00:\($0)")}
.disposed(by: disposeBag)
let netOB = Observable<Any>.create { (observer) -> Disposable in
sleep(2)// 模拟网络延迟
print("我开始请求网络了")
observer.onNext("请求到的网络数据")
observer.onNext("请求到的本地")
observer.onCompleted()
return Disposables.create {
print("销毁回调了")
}
}.publish()
netOB.subscribe(onNext: { (anything) in
print("订阅1:",anything)
})
.disposed(by: disposeBag)
// 我们有时候不止一次网络订阅,因为有时候我们的数据可能用在不同的额地方
// 所以在订阅一次 会出现什么问题?
netOB.subscribe(onNext: { (anything) in
print("订阅2:",anything)
})
.disposed(by: disposeBag)
_ = netOB.connect()
}
- 底层逻辑探索中间变量
ConnectableObservableAdapter
保存了源序列source
、中间序列makeSubject
- 订阅流程
self.lazySubject.subscribe(observer)
一个懒加载的序列,保证了中间变量ConnectableObservableAdapter
每一次都是同一个响应序列 - 剩下就是
PublishSubject
的订阅效果 - 完事等待源序列的响应,但是我们的源序列的订阅是在
connect
函数里面!如果没有调用connect
函数,意味着就永远不会发送响应。这样背后的逻辑就是,前面所以的发送响应在connect
函数之前的都没有任何的意义! - 以上也就说明了我们的
publish
就是状态共享的:connnect
一次我们序列发送一次响应(响应所有订阅)。
7.2:replay
- 将源可观察序列转换为可连接的序列,并将向每个新订阅服务器重放以前排放的缓冲大小
- 首先拥有和
publish
一样的能力,共享Observable sequence
, 其次使用replay
还需要我们传入一个参数(buffer size)
来缓存已发送的事件,当有新的订阅者订阅了,会把缓存的事件发送给新的订阅者
func testReplayConnectOperators(){
print("*****replay*****")
let interval = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).replay(5)
interval.subscribe(onNext: { print(Date.time,"订阅: 1, 事件: \($0)") })
.disposed(by: self.disposeBag)
delay(2) { _ = interval.connect() }
delay(4) {
interval.subscribe(onNext: { print(Date.time,"订阅: 2, 事件: \($0)") })
.disposed(by: self.disposeBag)
}
delay(8) {
interval.subscribe(onNext: { print(Date.time,"订阅: 3, 事件: \($0)") })
.disposed(by: self.disposeBag)
}
delay(20, closure: {
self.disposeBag = DisposeBag()
})
/**
订阅: 1, 事件: 4
订阅: 1, 事件: 0
2019-05-28 21-32-42 订阅: 2, 事件: 0
2019-05-28 21-32-42 订阅: 1, 事件: 1
2019-05-28 21-32-42 订阅: 2, 事件: 1
2019-05-28 21-32-45 订阅: 2, 事件: 4
2019-05-28 21-32-46 订阅: 3, 事件: 0
2019-05-28 21-32-46 订阅: 3, 事件: 1
2019-05-28 21-32-46 订阅: 3, 事件: 2
2019-05-28 21-32-46 订阅: 3, 事件: 3
2019-05-28 21-32-46 订阅: 3, 事件: 4
// 序列从 0开始
// 定时器也没有断层 sub2 sub3 和 sub1 是同步的
*/
}
高阶函数的用法还是有点意思的!如果你是一个
RxSwift
新手,还是非常建议大家耐着性子好好练习,对你的开发绝对有帮助!就问此时此刻还有谁?45度仰望天空,该死!我这无处安放的魅力!