RxSwift学习之旅-Operators
2017-05-25 本文已影响252人
ripple_k
Operators是RxSwift的基本组成部分，它的重要性就不言而喻了，前面只是提到在我们开发中，当收到observable发出.next的事件时，如果其携带的value并不是我们想要直接拿来使用的那个value，那我们就需要在我们拿到最终的value之前，对数据进行相应的处理
Filtering Operators
Ignoring operators
ignoreElements
ignoreElements能够忽略.next的事件，当我们只关心observable sequence是否结束时使用
example(of: "ignoreElements") {
let strikes = PublishSubject<String>()
strikes.ignoreElements().subscribe({ (event) in
print(event)
}).disposed(by: dispose)
strikes.onNext("X")
strikes.onNext("X")
strikes.onCompleted()
// strikes.onError(MyError.myError)
}
--- Example of: ignoreElements ---
completed
上述事例中，我们使用ignoreElements忽略了两次.onNext("X")事件，只有当stries发出onCompleted或onError事件时，我们的订阅才能接收到响应，整个过程如下图所示
Snip20170525_2.png
elementAt
有些时候你只关注第n个元素发出的事件，而忽略其它事件elementAt就派上用场了
example(of: "elementAt") {
let strikes = PublishSubject<String>()
strikes.elementAt(2).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
strikes.onNext("1")
strikes.onNext("2")
strikes.onNext("3")
strikes.onNext("4")
}
--- Example of: elementAt ---
next(3)
completed
需要注意的是，上述事例中，我们并没有发出completed事件，但是当observable sequence响应到指定下下标元素的.onNext事件后，observable sequence会主动结束。提醒一下，这个下标是从0开始的哦~
那如果我们提前发送了completed事件会怎样呢
example(of: "elementAt") {
let strikes = PublishSubject<String>()
strikes.elementAt(2).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
strikes.onNext("1")
strikes.onNext("2")
strikes.onCompleted()
strikes.onNext("3")
strikes.onNext("4")
}
--- Example of: elementAt ---
error(Argument out of range.)
当订阅还没有响应到指定下标的事件而提前发出.onCompleted()的话，这时候响应的将是一个error。整个过程如下图所示
Snip20170525_3.png
filter
filter只允许满足条件的value通过
example(of: "filter") {
Observable.of(1,2,3,4,5,6,7,8,9,10).filter({ (value) -> Bool in
value % 2 == 0
}).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
}
--- Example of: filter ---
next(2)
next(4)
next(6)
next(8)
next(10)
completed
过程如下图
Snip20170525_5.png
Skipping operators
skip
skip跳过前几个
example(of: "skip") {
Observable.of("A", "B", "C", "D", "E", "F", "G").skip(3).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
}
--- Example of: skip ---
next(D)
next(E)
next(F)
next(G)
completed
如图
Snip20170525_6.png
skipWhile
skipWhile一旦有元素让通过，后面的元素都被允许通过
example(of: "skipWhile") {
Observable.of(1,2,3,4,5,6,7,8,9,10,2,3,4).skipWhile({ (value) -> Bool in
return value < 5
}).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
}
--- Example of: skipWhile ---
next(5)
next(6)
next(7)
next(8)
next(9)
next(10)
next(2)
next(3)
next(4)
completed
如图
Snip20170525_8.png
skipUntil
skipUntil(Observable)直到Observable发出.next事件，才允许通过
example(of: "skipUntil") {
let subject = PublishSubject<String>()
let trigger = PublishSubject<String>()
subject.skipUntil(trigger).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
subject.onNext("A")
subject.onNext("B")
trigger.onNext("aaa")
subject.onNext("C")
subject.onNext("D")
}
--- Example of: skipUntil ---
next(C)
next(D)
直到trigger发出.onNext("aaa")事件，subject才能响应事件，如图
Snip20170525_9.png
Taking operators
take
take(n)只取前n个值。取完后自己completed
example(of: "take") {
Observable.of(1,2,3,4,5,6,7,8,9).take(3).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
}
--- Example of: take ---
next(1)
next(2)
next(3)
completed
如果n>elememt的个数，则全部取完后结束，如图
Snip20170525_10.png
takeWhileWithIndex
takeWhileWithIndex从第一个元素开始取，takeWhileWithIndex闭包的两个参数分别代表序列中的元素和下标，takeWhileWithIndex闭包中的条件只要不成立则终止取值
example(of: "takeWhileWithIndex") {
Observable.of(1,2,3,4,3,2,7,8,9).takeWhileWithIndex({ (element, index) -> Bool in
return element < 5 && index < 3
}).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
}
--- Example of: takeWhileWithIndex ---
next(1)
next(2)
next(3)
completed
takeUntil
takeUntil(Observable)只要Observable发出.next事件，则序列终止
example(of: "takeUntil") {
let subject = PublishSubject<String>()
let trigger = PublishSubject<String>()
subject.takeUntil(trigger).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
subject.onNext("1")
subject.onNext("2")
trigger.onNext("X")
subject.onNext("3")
}
--- Example of: takeUntil ---
next(1)
next(2)
completed
trigger发出.onNext("X")事件，subject终止响应事件，如图
Snip20170525_11.png
takeUntil在RxCocoa中有一个很好的应用场景
someObservable.takeUntil(self.rx.deallocated) .subscribe(onNext: {
print($0)
})
Distinct operators
distinctUntilChanged
distinctUntilChanged如果连续的value相同，则被忽略，直到发生改变
example(of: "distinctUntilChanged") {
Observable.of("A", "A", "B", "B", "A").distinctUntilChanged()
.subscribe(onNext: {
print($0)
}).addDisposableTo(dispose)
}
--- Example of: distinctUntilChanged ---
A
B
A
如图
Snip20170525_12.png
distinctUntilChanged(comparer: @escaping (Self.E, Self.E) throws -> Bool)
自定义一个比较器，只有符合比较器条件的值才能通过
example(of: "distinctUntilChanged(comparer: @escaping (Self.E, Self.E) throws -> Bool)") {
Observable.of(1,2,3,10,10,5,6,70,8,9).distinctUntilChanged({ (element1, element2) -> Bool in
print("element1:", element1, "element2:", element2)
return element1 > element2
}).subscribe({ (event) in
print(event)
}).disposed(by: dispose)
}
--- Example of: distinctUntilChanged(comparer: @escaping (Self.E, Self.E) throws -> Bool) ---
next(1)
element1: 1 element2: 2
next(2)
element1: 2 element2: 3
next(3)
element1: 3 element2: 10
next(10)
element1: 10 element2: 10
next(10)
element1: 10 element2: 5
element1: 10 element2: 6
element1: 10 element2: 70
next(70)
element1: 70 element2: 8
element1: 70 element2: 9
completed
上面例子中，自定义比较器element1 > element2，条件不成立，则element1取较大的值，并发射，element2依次往后取与element1进行比较。原理如图
Paste_Image.png
Transforming Operators
toArray
将序列中的若干元素转换为一个数组元素
example(of: "toArray") {
let disposeBag = DisposeBag()
Observable.of("A", "B", "C").toArray().subscribe({ (event) in
print(event)
}).disposed(by: disposeBag)
}
--- Example of: toArray ---
next(["A", "B", "C"])
completed
Snip20170601_1.png
map
将sequence中的值进行自定义转换
example(of: "map") {
let disposeBag = DisposeBag()
Observable<Int>.of(1,2,3).map({
$0 * 2
}).subscribe(onNext: { (value) in
print(value)
}).disposed(by: disposeBag)
}
--- Example of: map ---
2
4
6
Snip20170601_2.png
mapWithIndex
example(of: "mapWithIndex") {
let disposeBag = DisposeBag()
Observable.of(1,2,3,4,5,6).mapWithIndex({ (integer, index) in
index > 2 ? integer * 2 : integer
}).subscribe({
print($0)
}).disposed(by: disposeBag)
}
--- Example of: mapWithIndex ---
next(1)
next(2)
next(3)
next(8)
next(10)
next(12)
completed
Snip20170601_3.png
flatMap
将一个序列发射的值转换成序列，然后将他们压平到一个序列。这也类似于 SequenceType 中的 flatMap。比如说我想将Observable<Student> 的序列转换为 Observable<Int> 序列
example(of: "flatMap") {
struct Student {
var score: Variable<Int>
}
let disposeBag = DisposeBag()
let xiaoMing = Student(score: Variable(80))
let xiaoHong = Student(score: Variable(95))
let student = PublishSubject<Student>()
student.asObservable().flatMap({
$0.score.asObservable()
}).subscribe({
print($0)
}).disposed(by: disposeBag)
student.onNext(xiaoHong)
student.onNext(xiaoMing)
xiaoMing.score.value = 100
}
--- Example of: flatMap ---
next(95)
next(80)
next(100)
Snip20170601_4.png
flatMapLatest
跟flatMap相比较，flatMapLatest只能响应最后一个元素值的变化
example(of: "flatMapLatest") {
struct Student {
var score: Variable<Int>
}
let disposeBag = DisposeBag()
let xiaoMing = Student(score: Variable(80))
let xiaoHong = Student(score: Variable(95))
let student = PublishSubject<Student>()
student.asObservable().flatMapLatest({
$0.score.asObservable()
}).subscribe({
print($0)
}).disposed(by: disposeBag)
student.onNext(xiaoMing)
xiaoMing.score.value = 65
student.onNext(xiaoHong)
xiaoMing.score.value = 88
xiaoHong.score.value = 100
}
--- Example of: flatMapLatest ---
next(80)
next(65)
next(95)
next(100)
Snip20170601_5.png
Combining Operators
startWith
指定队列起始元素
example(of: "startWith") {
let numbers = Observable.of(2,3,4)
let observable = numbers.startWith(1)
observable.subscribe(onNext: { (element) in
print(element)
}).dispose()
}
--- Example of: startWith ---
1
2
3
4
Snip20170601_6.png
Observable.concat
合并两个队列，concat合并的两个sequence元素必须是相同类型
example(of: "Observable.concat") {
let first = Observable.of(1,2,3)
let second = Observable.of(4,5,6)
let observable = Observable.concat([first, second])
observable.subscribe(onNext: { (element) in
print(element)
}).dispose()
}
--- Example of: Observable.concat ---
1
2
3
4
5
6
concat
example(of: "concat") {
let germanCities = Observable.of("Berlin", "Münich", "Frankfurt")
let spanishCities = Observable.of("Madrid", "Barcelona", "Valencia")
let observable = germanCities.concat(spanishCities)
observable.subscribe(onNext: { value in
print(value)
}).dispose()
}
--- Example of: concat ---
Berlin
Münich
Frankfurt
Madrid
Barcelona
Valencia
concat one element
example(of: "concat one element") {
let numbers = Observable.of(2,3,4)
let observable = Observable.just(1).concat(numbers)
observable.subscribe(onNext: { (element) in
print(element)
}).dispose()
}
--- Example of: concat one element ---
1
2
3
4
Snip20170601_7.png
merge
把两个sequence合并成一个
example(of: "merge") {
let left = PublishSubject<String>()
let right = PublishSubject<String>()
let source = Observable.of(left.asObservable(), right.asObservable())
let observable = source.merge()
let disposable = observable.subscribe(onNext: { value in
print(value)
})
var leftValues = ["Berlin", "Munich", "Frankfurt"]
var rightValues = ["Madrid", "Barcelona", "Valencia"]
repeat {
if arc4random_uniform(2) == 0 {
if !leftValues.isEmpty {
left.onNext("Left: " + leftValues.removeFirst())
}
} else if !rightValues.isEmpty {
right.onNext("Right: " + rightValues.removeFirst())
}
} while !leftValues.isEmpty || !rightValues.isEmpty
disposable.dispose()
}
--- Example of: merge ---
Right: Madrid
Right: Barcelona
Left: Berlin
Right: Valencia
Left: Munich
Left: Frankfurt
Snip20170601_8.png
combineLatest
将两个observable绑定，只有当它们都发出事件的时候才能subscribe到它们的最新发送的值
example(of: "combineLatest") {
let left = PublishSubject<String>()
let right = PublishSubject<String>()
let observable = Observable.combineLatest(left,right)
let disposable = observable.subscribe(onNext: { (left,right) in
print(left,right)
})
print("> Sending a value to Left")
left.onNext("Hello,")
print("> Sending a value to Right")
right.onNext("world")
print("> Sending another value to Right")
right.onNext("RxSwift")
print("> Sending another value to Left")
left.onNext("Have a good day,")
disposable.dispose()
}
--- Example of: combineLatest ---
> Sending a value to Left
> Sending a value to Right
Hello, world
> Sending another value to Right
Hello, RxSwift
> Sending another value to Left
Have a good day, RxSwift
example(of: "combine user choice and value") {
let choice : Observable<DateFormatter.Style> =
Observable.of(.short, .long)
let dates = Observable.of(Date())
let observable = Observable.combineLatest(choice, dates) {
(format, when) -> String in
let formatter = DateFormatter()
formatter.dateStyle = format
return formatter.string(from: when)
}
observable.subscribe(onNext: { value in
print(value)
})
}
--- Example of: combine user choice and value ---
6/1/17
June 1, 2017
Snip20170601_9.png
zip
把多个sequence合并为一个sequence，当所有的observable都产生element的时候，selector function会生成一个对应索引的element
example(of: "zip") {
enum Weather {
case cloudy
case sunny
}
let bag = DisposeBag()
let left = Observable<Weather>.of(.sunny, .cloudy, .cloudy, .sunny)
let right = Observable.of("Lisbon", "Copenhagen", "London", "Madrid", "Vienna")
let observable = Observable.zip(left, right){ (weather, city) -> String in
"it's \(weather) in \(city)"
}
observable.subscribe(onNext: { (result) in
print(result)
}).addDisposableTo(bag)
}
--- Example of: zip ---
it's sunny in Lisbon
it's cloudy in Copenhagen
it's cloudy in London
it's sunny in Madrid
Snip20170602_10.png
Triggers
withLatestFrom
将两个 observable sequence 合并到一个 observable sequence ，每当self发出element将获取到第二个 observable sequence 的最新element
example(of: "withLatestFrom") {
let bag = DisposeBag()
let button = PublishSubject<Void>()
let textField = PublishSubject<String>()
let observable = button.withLatestFrom(textField)
observable.subscribe(onNext: { (text) in
print(text)
}).addDisposableTo(bag)
textField.onNext("Par")
textField.onNext("Pari")
textField.onNext("Paris")
button.onNext()
button.onNext()
}
--- Example of: withLatestFrom ---
Paris
Paris
Snip20170602_11.png
sample
每当sampler产生一个
.next()事件，将从源sequence中获取最新element，如果在sampler两次事件间隔中源sequence没有新element产生，则获取不到element
example(of: "sample") {
let bag = DisposeBag()
let button = PublishSubject<Void>()
let textField = PublishSubject<String>()
let observable = textField.sample(button)
observable.subscribe(onNext: { (text) in
print(text)
}).addDisposableTo(bag)
textField.onNext("Par")
textField.onNext("Pari")
textField.onNext("Paris")
button.onNext()
button.onNext()
}
--- Example of: sample ---
Paris
Snip20170602_12.png
Switches
amb
两个observable sequence谁先发送element谁就是第一响应者。
example(of: "amb") {
let left = PublishSubject<String>()
let right = PublishSubject<String>()
let observable = left.amb(right)
let disposable = observable.subscribe(onNext: { value in
print(value)
})
right.onNext("Copenhagen")
left.onNext("Lisbon")
left.onNext("London")
left.onNext("Madrid")
right.onNext("Vienna")
disposable.dispose()
}
--- Example of: amb ---
Copenhagen
Vienna
Snip20170602_13.png
switchLatest
响应最新的observable sequence中的element
example(of: "switchLatest") {
let one = PublishSubject<String>()
let two = PublishSubject<String>()
let three = PublishSubject<String>()
let source = PublishSubject<Observable<String>>()
let observable = source.switchLatest()
let dispose = observable.subscribe(onNext: { (value) in
print(value)
})
source.onNext(one)
one.onNext("one first")
two.onNext("two first")
source.onNext(two)
one.onNext("one second")
two.onNext("two second")
source.onNext(three)
three.onNext("three first")
one.onNext("one third")
three.onNext("three second")
source.onNext(two)
two.onNext("two third")
three.onNext("three third")
}
--- Example of: switchLatest ---
one first
two second
three first
three second
two third
Snip20170602_14.png
reduce
将一个observable sequence中的元素做一个累加，返回一个单一的结果。第一个参数为最初的累加值
example(of: "reduce") {
let source = Observable.of(1, 3, 5, 7, 9)
let observable = source.reduce(0, accumulator: { summary, newValue in
return summary + newValue
})
// 可简写为
// let observable = source.reduce(0, accumulator: +)
observable.subscribe(onNext: { value in
print(value)
})
}
--- Example of: reduce ---
25
Snip20170602_15.png
scan
将一个observable sequence中的元素做一个累加，返回每个累加结果。第一个参数为最初的累加值
example(of: "scan") {
let source = Observable.of(1,3,5,7,9)
let observable = source.scan(0, accumulator: { (summary, newValue) -> Int in
return summary + newValue
})
// 可简写为
// let observable = source.scan(0, accumulator: +)
observable.subscribe(onNext: { (value) in
print(value)
})
}
--- Example of: scan ---
1
4
9
16
25
Snip20170602_16.png
小结
最后提一下RxSwift所有的Operators都是定义在ObservableType的协议延展里面的，返回值都是 -> RxSwift.Observable<Self.E>，以满足开发者在日常开发中的链式调用，所以在日常开发中，你可以这样写代码
input.skipWhile { $0 == 0 }
.filter { $0 < 10 }
.take(10)
.toArray()
.subscribe(onNext: {
print($0)
})
.addDisposableTo(disposeBag)
