Introduction:
Creating and Subscribing to Observables:
There are several ways to create and subscribe to Observable sequences.
never:
Creates a sequence that never terminates and never emits any events.more info
let disposeBag = DisposeBag()
let neverSequence = Observable<String>.never()
let neverSequenceSubscription = neverSequence
.subscribe { _ in
print("This will never be printed")
}
neverSequenceSubscription.disposed(by: disposeBag)
empty:
Creates an empty Observable sequence that only emits a Completed event. [More info](Creates an empty Observable sequence that only emits a Completed event. More info)
let disposeBag = DisposeBag()
Observable<Int>.empty()
.subscribe { event in
print(event)
}
.disposed(by: disposeBag)
just:
Creates an Observable sequence with a single element. More info
let disposeBag = DisposeBag()
Observable.just("🔴")
.subscribe { event in
print(event)
}
.disposed(by: disposeBag)
of:
Creates an Observable sequence with a fixed number of elements.
let disposeBag = DisposeBag()
Observable.of("🐶", "🐱", "🐭", "🐹")
.subscribe(onNext: { element in
print(element)
})
.disposed(by: disposeBag)
from:
Creates an Observable sequence from a Sequence, such as an Array, Dictionary, or Set.
let disposeBag = DisposeBag()
Observable.from(["🐶", "🐱", "🐭", "🐹"])
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
create:
Creates a custom Observable sequence. More info
let disposeBag = DisposeBag()
let myJust = { (element: String) -> Observable<String> in
return Observable.create { observer in
observer.on(.next(element))
observer.on(.completed)
return Disposables.create()
}
}
myJust("🔴")
.subscribe { print($0) }
.disposed(by: disposeBag)
range:
Creates an Observable sequence that emits a range of sequential integers and then terminates. More info
let disposeBag = DisposeBag()
Observable.range(start: 1, count: 10)
.subscribe { print($0) }
.disposed(by: disposeBag)
repeatElement:
Creates an Observable sequence that emits the given element indefinitely. More info
let disposeBag = DisposeBag()
Observable.repeatElement("🔴")
.take(3)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
generate:
Creates an Observable sequence that generates values for as long as the provided condition evaluates to true.
let disposeBag = DisposeBag()
Observable.generate(
initialState: 0,
condition: { $0 < 3 },
iterate: { $0 + 1 }
)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
deferred:
Creates a new Observable sequence for each subscriber. More info
let disposeBag = DisposeBag()
var count = 1
let deferredSequence = Observable<String>.deferred {
print("Creating \(count)")
count += 1
return Observable.create { observer in
print("Emitting...")
observer.onNext("🐶")
observer.onNext("🐱")
observer.onNext("🐵")
return Disposables.create()
}
}
deferredSequence
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
deferredSequence
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
error:
Creates an Observable sequence that emits no items and immediately terminates with an error.
let disposeBag = DisposeBag()
Observable<Int>.error(TestError.test)
.subscribe { print($0) }
.disposed(by: disposeBag)
doOn:
Invokes a side-effect action for each emitted event and returns (passes through) the original event. More info
let disposeBag = DisposeBag()
Observable.of("🍎", "🍐", "🍊", "🍋")
.do(onNext: { print("Intercepted:", $0) }, onError: { print("Intercepted error:", $0) }, onCompleted: { print("Completed") })
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
Working with Subjects:
A Subject is a sort of bridge or proxy that is available in some implementations of Rx that acts as both an observer and Observable. Because it is an observer, it can subscribe to one or more Observables, and because it is an Observable, it can pass through the items it observes by reemitting them, and it can also emit new items. More info
一个Subject是一种桥梁或代理,在Rx的一些实现中可用作观察者和Observable。因为它是一个观察者,所以可以订阅一个或多个Observable,由于它是一个Observable,它可以通过重新发送它们来通过它观察到的项目,也可以发出新的项目。
/**
Add observer with `id` and print each emitted event.
- parameter id: an identifier for the subscription.
*/
func addObserver(_ id: String) -> Disposable {
return subscribe { print("Subscription:", id, "Event:", $0) }
}
PublishSubject:
Broadcasts new events to all observers as of their time of the subscription.
在观察者订阅的时间向它们广播新的事件。
let disposeBag = DisposeBag()
let subject = PublishSubject<String>()
subject.addObserver("1").disposed(by: disposeBag)
subject.onNext("🐶")
subject.onNext("🐱")
subject.addObserver("2").disposed(by: disposeBag)
subject.onNext("🅰️")
subject.onNext("🅱️")
Subscription: 1 Event: next(🐶)
Subscription: 1 Event: next(🐱)
Subscription: 1 Event: next(🅰️)
Subscription: 2 Event: next(🅰️)
Subscription: 1 Event: next(🅱️)
Subscription: 2 Event: next(🅱️)
ReplaySubject:
Broadcasts new events to all subscribers, and the specified bufferSize number of previous events to new subscribers.
向所有订阅者广播新事件,并向新订阅者广播指定以前事件序列中缓冲区大小数量的事件(从后往前数)。
let disposeBag = DisposeBag()
let subject = ReplaySubject<String>.create(bufferSize: 1)
subject.addObserver("1").disposed(by: disposeBag)
subject.onNext("🐶")
subject.onNext("🐱")
subject.addObserver("2").disposed(by: disposeBag)
subject.onNext("🅰️")
subject.onNext("🅱️")
Subscription: 1 Event: next(🐶)
Subscription: 1 Event: next(🐱)
Subscription: 2 Event: next(🐱)
Subscription: 1 Event: next(🅰️)
Subscription: 2 Event: next(🅰️)
Subscription: 1 Event: next(🅱️)
Subscription: 2 Event: next(🅱️)
BehaviorSubject:
Broadcasts new events to all subscribers, and the most recent (or initial) value to new subscribers.
向所有用户广播新的事件,并向新的用户提供最近(或最初)的价值。
let disposeBag = DisposeBag()
let subject = BehaviorSubject(value: "🔴")
subject.addObserver("1").disposed(by: disposeBag)
subject.onNext("🐶")
subject.onNext("🐱")
subject.addObserver("2").disposed(by: disposeBag)
subject.onNext("🅰️")
subject.onNext("🅱️")
subject.addObserver("3").disposed(by: disposeBag)
subject.onNext("🍐")
subject.onNext("🍊")
Subscription: 1 Event: next(🔴)
Subscription: 1 Event: next(🐶)
Subscription: 1 Event: next(🐱)
Subscription: 2 Event: next(🐱)
Subscription: 1 Event: next(🅰️)
Subscription: 2 Event: next(🅰️)
Subscription: 1 Event: next(🅱️)
Subscription: 2 Event: next(🅱️)
Subscription: 3 Event: next(🅱️)
Subscription: 1 Event: next(🍐)
Subscription: 2 Event: next(🍐)
Subscription: 3 Event: next(🍐)
Subscription: 1 Event: next(🍊)
Subscription: 2 Event: next(🍊)
Subscription: 3 Event: next(🍊)
Variable:
Wraps a BehaviorSubject, so it will emit the most recent (or initial) value to new subscribers. And Variable also maintains current value state. Variable will never emit an Error event. However, it will automatically emit a Completed event and terminate on deinit.
包装一个BehaviorSubject,所以它会发送最新的(或初始)值给新的观察者。Variable也保持当前值状态。Variable将永远不会发出一个错误事件。但是,它会自动发出一个Completed事件,并在deinit终止。
let disposeBag = DisposeBag()
let variable = Variable("🔴")
variable.asObservable().addObserver("1").disposed(by: disposeBag)
variable.value = "🐶"
variable.value = "🐱"
variable.asObservable().addObserver("2").disposed(by: disposeBag)
variable.value = "🅰️"
variable.value = "🅱️"
Subscription: 1 Event: next(🔴)
Subscription: 1 Event: next(🐶)
Subscription: 1 Event: next(🐱)
Subscription: 2 Event: next(🐱)
Subscription: 1 Event: next(🅰️)
Subscription: 2 Event: next(🅰️)
Subscription: 1 Event: next(🅱️)
Subscription: 2 Event: next(🅱️)
Subscription: 1 Event: completed
Subscription: 2 Event: completed
Combining_Operators 联合操作符
联合多个Observable到一个Observable
startWith:
Emits the specified sequence of elements before beginning to emit the elements from the source Observable. More info
在开始发送原始的元素序列之前,发射指定的元素序列。
你可以链式的对一个Observable调用多次startWith,但是发射的序列是按后进先出(last-in-first-out)的方式,即后startWith的元素序列会先发射出去
let disposeBag = DisposeBag()
Observable.of("🐶", "🐱", "🐭", "🐹")
.startWith("1️⃣")
.startWith("2️⃣")
.startWith("3️⃣", "🅰️", "🅱️")
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
3️⃣
🅰️
🅱️
2️⃣
1️⃣
🐶
🐱
🐭
🐹
merge:
Combines elements from source Observable sequences into a single new Observable sequence, and will emit each element as it is emitted by each source Observable sequence. More info
结合多个发射序列为一个新的序列
let disposeBag = DisposeBag()
let subject1 = PublishSubject<String>()
let subject2 = PublishSubject<String>()
Observable.of(subject1, subject2)
.merge()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
subject1.onNext("🅰️")
subject1.onNext("🅱️")
subject2.onNext("①")
subject2.onNext("②")
subject1.onNext("🆎")
subject2.onNext("③")
🅰️
🅱️
①
②
🆎
③
zip:
Combines up to 8 source Observable sequences into a single new Observable sequence, and will emit from the combined Observable sequence the elements from each of the source Observable sequences at the corresponding index. More info
将最多8个源“Observable”序列组合成一个新的“Observable”序列,并且将从组合的“Observable”序列发出来自相应索引处的每个源“Observable”序列的元素。发射的序列个数是根据最少的序列个数来定的
let stringSubject = PublishSubject<String>()
let intSubject = PublishSubject<Int>()
let intSubject2 = PublishSubject<Int>()
Observable.zip(stringSubject, intSubject, intSubject2) { stringElement, intElement, int2 in
"\(stringElement) \(intElement) \(int2)"
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
stringSubject.onNext("🅰️")
stringSubject.onNext("🅱️")
intSubject2.onNext(3)
intSubject2.onNext(5)
intSubject.onNext(1)
intSubject.onNext(2)
stringSubject.onNext("🆎")
intSubject.onNext(3)
🅰️ 1 3
🅱️ 2 5
combineLatest:
Combines up to 8 source Observable sequences into a single new Observable sequence, and will begin emitting from the combined Observable sequence the latest elements of each source Observable sequence once all source sequences have emitted at least one element, and also when any of the source Observable sequences emits a new element. More info
将最多8个源Observable序列组合成一个新的Observable序列,并且一旦所有源序列发射至少一个元素,将从组合的Observable序列开始发射每个源“可观察”序列的最新元素,还有当任何源Observable
序列发出一个新的元素时候,也从其它Observable中拿出最近的元素来组合发射
let disposeBag = DisposeBag()
let stringSubject = PublishSubject<String>()
let intSubject = PublishSubject<Int>()
Observable.combineLatest(stringSubject, intSubject) { stringElement, intElement in
"\(stringElement) haha \(intElement)"
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
stringSubject.onNext("🅰️")
stringSubject.onNext("🅱️")
intSubject.onNext(1)
intSubject.onNext(2)
stringSubject.onNext("🆎")
🅱️ haha 1
🅱️ haha 2
🆎 haha 2
Array.combineLatest 和combineLatest类似
let disposeBag = DisposeBag()
let stringObservable = Observable.just("❤️")
let fruitObservable = Observable.from(["🍎", "🍐", "🍊"])
let animalObservable = Observable.of("🐶", "🐱", "🐭", "🐹")
Observable.combineLatest([stringObservable, fruitObservable, animalObservable]) {
"\($0[0]) \($0[1]) \($0[2])"
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
❤️ 🍎 🐶
❤️ 🍐 🐶
❤️ 🍐 🐱
❤️ 🍊 🐱
❤️ 🍊 🐭
❤️ 🍊 🐹
switchLatest:
Transforms the elements emitted by an Observable sequence into Observable sequences, and emits elements from the most recent inner Observable sequence. More info
将由观察序列发出的元素转换为观察序列,并且只有最新的观察序列能发送信号
let disposeBag = DisposeBag()
let subject1 = BehaviorSubject(value: "⚽️")
let subject2 = BehaviorSubject(value: "🍎")
let variable = Variable(subject1)
variable.asObservable()
.switchLatest()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
subject1.onNext("🏈")
subject1.onNext("🏀")
variable.value = subject2
//此时不会打印⚾️因为当前最新的variable是subject2
subject1.onNext("⚾️")
subject2.onNext("🍐")
⚽️
🏈
🏀
🍎
🍐
Transforming_Operators:
Operators that transform Next event elements emitted by an Observable
sequence.
转换操作符:转换发射的Next事件元素到一个新的Observable序列
map:
Applies a transforming closure to elements emitted by an Observable
sequence, and returns a new Observable
sequence of the transformed elements.
将变换闭包应用于由“Observable”序列发出的元素,并返回变换元素的一个新的“Observable”序列。
example("map") {
let disposeBag = DisposeBag()
Observable.of(1, 2, 3)
.map { $0 * $0 }
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
}
1
4
9
flatMap and flatMapLatest:
Transforms the elements emitted by an Observable
sequence into Observable
sequences, and merges the emissions from both Observable
sequences into a single Observable
sequence. This is also useful when, for example, when you have an Observable
sequence that itself emits Observable
sequences, and you want to be able to react to new emissions from either Observable
sequence. The difference between flatMap
and flatMapLatest
is, flatMapLatest
will only emit elements from the most recent inner Observable
sequence.
将由“可观察”序列发射的元素转换为“可观察”序列,并将来自两个“可观察”序列的发射合并成单个“可观察”序列。例如,当你有一个本身发出“可观察”序列的“可观察”序列,并且你希望能够对来自“可观察”序列的新发射作出反应时,这也是有用的。 “flatMap”和“flatMapLatest”的区别在于,“flatMapLatest”只会从最近的“Observable”内部发出元素。
flatMapLatest是map和switchLatest的结合体。
let disposeBag = DisposeBag()
struct Player {
var score: Variable<Int>
}
let 👦🏻 = Player(score: Variable(80))
let 👧🏼 = Player(score: Variable(90))
let player = Variable(👦🏻)
player.asObservable()
.flatMapLatest { $0.score.asObservable() } // Change flatMap to flatMapLatest and observe change in printed output
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
👦🏻.score.value = 85
player.value = 👧🏼
👦🏻.score.value = 95 // Will be printed when using flatMap, but will not be printed when using flatMapLatest
👧🏼.score.value = 100
80
85
90
100
scan:
Begins with an initial seed value, and then applies an accumulator closure to each element emitted by an Observable
sequence, and returns each intermediate result as a single-element Observable
sequence.
从一个初始的种子值开始,然后对由“可观察”序列发出的每个元素应用累加器闭包,并将每个中间结果作为单元素“可观察”序列返回。
let disposeBag = DisposeBag()
Observable.of(10, 100, 1000)
.scan(1) { aggregateValue, newValue in
aggregateValue + newValue
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
11
111
1111
Filtering and Conditional Operators:
Operators that selectively emit elements from a source Observable
sequence.
有选择地从源“Observable”序列中发射元素的运算符
filter:
Emits only those elements from an Observable
sequence that meet the specified condition.
仅从符合指定条件的“可观察”序列中发射那些元素
let disposeBag = DisposeBag()
Observable.of(
"🐱", "🐰", "🐶",
"🐸", "🐱", "🐰",
"🐹", "🐸", "🐱")
.filter {
$0 == "🐱"
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐱
🐱
🐱
distinctUntilChanged:
Suppresses sequential duplicate elements emitted by an Observable
sequence.More info
抑制由“可观察”序列发出的连续重复元素。
[图片上传失败...(image-c60fb6-1515048671653)]
let disposeBag = DisposeBag()
Observable.of("🐱", "🐷", "🐱", "🐱", "🐱", "🐵", "🐱")
.distinctUntilChanged()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐱
🐷
🐱
🐵
🐱
elementAt:
Emits only the element at the specified index of all elements emitted by an Observable
sequence. More info
仅发出由“Observable”序列发出的所有元素的指定索引处的元素
[图片上传失败...(image-4979d3-1515048671653)]
let disposeBag = DisposeBag()
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.elementAt(3)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐸
single:
Emits only the first element (or the first element that meets a condition) emitted by an Observable
sequence. Will throw an error if the Observable
sequence does not emit exactly one element.
仅发出由“可观察”序列发出的第一个元素(或符合条件的第一个元素)。如果Observable
序列不是只发出一个元素,会抛出一个错误。
example("single") {
let disposeBag = DisposeBag()
Observable.of("🐱")
.single()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
}
example("single with conditions") {
let disposeBag = DisposeBag()
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.single { $0 == "🐸" }
.subscribe { print($0) }
.disposed(by: disposeBag)
Observable.of("🐱", "🐰", "🐶", "🐱", "🐰", "🐶")
.single { $0 == "🐰" }
.subscribe { print($0) }
.disposed(by: disposeBag)
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.single { $0 == "🔵" }
.subscribe { print($0) }
.disposed(by: disposeBag)
}
--- single example ---
🐱
--- single with conditions example ---
next(🐸)
completed
next(🐰)
error(Sequence contains more than one element.)
error(Sequence doesn't contain any elements.)
take:
Emits only the specified number of elements from the beginning of an Observable
sequence. More info
仅从“可观察”序列的开始处发射指定数量的元素。
[图片上传失败...(image-546204-1515048671653)]
let disposeBag = DisposeBag()
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.take(3)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐱
🐰
🐶
takeLast:
Emits only the specified number of elements from the end of an Observable
sequence. More info
仅从"可观察"序列的末尾处发射指定数量的元素。注意:发射元素的顺序不变,只不过是从后往前数,从第几个开始发射。
[图片上传失败...(image-789c87-1515048671653)]
let disposeBag = DisposeBag()
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.takeLast(3)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐸
🐷
🐵
takeWhile:
Emits elements from the beginning of an Observable
sequence as long as the specified condition evaluates to true
. More info
只要指定的条件评估为“真”,就从“可观察”序列的开始处发射元素。从开始往后,如果有元素不满足条件,将发射completed事件而不再发射元素。
[图片上传失败...(image-34022b-1515048671653)]
let disposeBag = DisposeBag()
Observable.of(1, 2, 3, 4, 5, 6)
.takeWhile { $0 % 2 == 0 }
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
1
2
3
takeUntil:
Emits elements from a source Observable
sequence until a reference Observable
sequence emits an element. More info
从源Observable序列中发射元素,直到参考Observable序列发射一个元素。
[图片上传失败...(image-7af76f-1515048671653)]
let disposeBag = DisposeBag()
let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()
sourceSequence
.takeUntil(referenceSequence)
.subscribe { print($0) }
.disposed(by: disposeBag)
sourceSequence.onNext("🐱")
sourceSequence.onNext("🐰")
sourceSequence.onNext("🐶")
referenceSequence.onNext("🔴")
sourceSequence.onNext("🐸")
sourceSequence.onNext("🐷")
sourceSequence.onNext("🐵")
next(🐱)
next(🐰)
next(🐶)
completed
skip:
Suppresses emitting the specified number of elements from the beginning of an Observable
sequence. More info
禁止从“可观察”序列的开始发射指定数量的元素。
[图片上传失败...(image-63e34f-1515048671653)]
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.skip(2)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐶
🐸
🐷
🐵
skipWhile:
Suppresses emitting the elements from the beginning of an Observable
sequence that meet the specified condition. More info
禁止从符合指定条件的“可观察”序列开始发射元素。
[图片上传失败...(image-8d937f-1515048671653)]
let disposeBag = DisposeBag()
Observable.of(1, 2, 3, 4, 5, 6)
.skipWhile { $0 < 4 }
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
4
5
6
skipWhileWithIndex:
Suppresses emitting the elements from the beginning of an Observable
sequence that meet the specified condition, and emits the remaining elements. The closure is also passed each element's index.
禁止从符合指定条件的“可观察”序列开始发射元素,并发射其余元素。闭包也传递每个元素的索引。
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
.skipWhileWithIndex { element, index in
index < 3
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🐸
🐷
🐵
skipUntil:
Suppresses emitting the elements from a source Observable
sequence until a reference Observable
sequence emits an element. More info
禁止从源“Observable”序列发射元素,直到引用“Observable”序列发射一个元素。
[图片上传失败...(image-301338-1515048671653)]
let disposeBag = DisposeBag()
let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()
sourceSequence
.skipUntil(referenceSequence)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
sourceSequence.onNext("🐱")
sourceSequence.onNext("🐰")
sourceSequence.onNext("🐶")
referenceSequence.onNext("🔴")
sourceSequence.onNext("🐸")
sourceSequence.onNext("🐷")
sourceSequence.onNext("🐵")
🐸
🐷
🐵
Mathematical and Aggregate Operators:
Operators that operate on the entire sequence of items emitted by an Observable
.
对Observable
发出的所有项目进行操作的操作符。
toArray:
Converts an Observable
sequence into an array, emits that array as a new single-element Observable
sequence, and then terminates. More info
将一个Observable序列转换为一个数组,将该数组作为一个新的单元素Observable序列发送,然后终止。
[图片上传失败...(image-4368b4-1515048671653)]
let disposeBag = DisposeBag()
Observable.range(start: 1, count: 10)
.toArray()
.subscribe { print($0) }
.disposed(by: disposeBag)
next([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
completed
reduce:
Begins with an initial seed value, and then applies an accumulator closure to all elements emitted by an Observable
sequence, and returns the aggregate result as a single-element Observable
sequence. More info
以一个初始的种子值开始,然后将一个累加器闭包应用于由一个“可观察”序列发出的所有元素,并且将该聚合结果作为单元素“可观察”序列返回。
[图片上传失败...(image-3c1124-1515048671653)]
let disposeBag = DisposeBag()
Observable.of(10, 100, 1000)
.reduce(1, accumulator: +)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
1111
concat:
Joins elements from inner Observable
sequences of an Observable
sequence in a sequential manner, waiting for each sequence to terminate successfully before emitting elements from the next sequence. More info
以顺序的方式连接来自“Observable”序列的内部“可观察”序列的元素,等待每个序列在从下一个序列发射元素之前成功终止。
[图片上传失败...(image-a9255f-1515048671653)]
let disposeBag = DisposeBag()
let subject1 = BehaviorSubject(value: "🍎")
let subject2 = BehaviorSubject(value: "🐶")
let variable = Variable(subject1)
variable.asObservable()
.concat()
.subscribe { print($0) }
.disposed(by: disposeBag)
subject1.onNext("🍐")
subject1.onNext("🍊")
variable.value = subject2
subject2.onNext("I would be ignored")
subject2.onNext("🐱")
subject1.onCompleted()
subject2.onNext("🐭")
next(🍎)
next(🍐)
next(🍊)
next(🐱)
next(🐭)
Connectable Operators:
Connectable Observable
sequences resembles ordinary Observable
sequences, except that they not begin emitting elements when subscribed to, but instead, only when their connect()
method is called. In this way, you can wait for all intended subscribers to subscribe to a connectable Observable
sequence before it begins emitting elements.
可连接的Observable
序列与普通的Observable
序列相似,只不过它们在订阅时不会开始发射元素,而只是在调用connect()方法的时候。通过这种方式,您可以等待所有预定用户在开始发射元素之前预订可连接的“可观察”序列。
interval:
interval
creates an Observable
sequence that emits elements after each period
, on the specified scheduler. More info
interval
在指定的调度器上创建一个Observable
序列,在每个`period“之后发出元素。
[图片上传失败...(image-237b29-1515048671653)]
let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
_ = interval
.subscribe(onNext: { print("Subscription: 1, Event: \($0)") })
delay(5) {
_ = interval
.subscribe(onNext: { print("Subscription: 2, Event: \($0)") })
}
Subscription: 1, Event: 0
Subscription: 1, Event: 1
Subscription: 1, Event: 2
Subscription: 1, Event: 3
Subscription: 1, Event: 4
Subscription: 1, Event: 5
Subscription: 2, Event: 0
Subscription: 1, Event: 6
Subscription: 2, Event: 1
Subscription: 1, Event: 7
Subscription: 2, Event: 2
Subscription: 1, Event: 8
Subscription: 2, Event: 3
Subscription: 1, Event: 9
Subscription: 2, Event: 4
Subscription: 1, Event: 10
Subscription: 2, Event: 5
Subscription: 1, Event: 11
publish:
Converts the source Observable
sequence into a connectable sequence. More info
将源“Observable”序列转换为可连接的序列。
[图片上传失败...(image-70b1a9-1515048671654)]
let intSequence = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
.publish()
_ = intSequence
.subscribe(onNext: { print("Subscription 1:, Event: \($0)") })
//如果我把这个connect延迟设为6的话,3个观察会一起打印
delay(2) { _ = intSequence.connect() }
delay(4) {
_ = intSequence
.subscribe(onNext: { print("Subscription 2:, Event: \($0)") })
}
delay(6) {
_ = intSequence
.subscribe(onNext: { print("Subscription 3:, Event: \($0)") })
}
Subscription 1:, Event: 0
Subscription 1:, Event: 1
Subscription 2:, Event: 1
Subscription 1:, Event: 2
Subscription 2:, Event: 2
Subscription 1:, Event: 3
Subscription 2:, Event: 3
Subscription 3:, Event: 3
Subscription 1:, Event: 4
Subscription 2:, Event: 4
Subscription 3:, Event: 4
Subscription 1:, Event: 5
Subscription 2:, Event: 5
Subscription 3:, Event: 5
Subscription 1:, Event: 6
Subscription 2:, Event: 6
Subscription 3:, Event: 6
Schedulers are an abstraction of mechanisms for performing work, such as on specific threads or dispatch queues. More info
replay:
Converts the source Observable
sequence into a connectable sequence, and will replay bufferSize
number of previous emissions to each new subscriber.More info
将源“Observable”序列转换为可连接的序列,并将向每个新订户播放先前发射的“bufferSize”数量。
[图片上传失败...(image-d7ede6-1515048671654)]
let intSequence = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
.replay(5)
_ = intSequence
.subscribe(onNext: { print("Subscription 1:, Event: \($0)") })
delay(2) { _ = intSequence.connect() }
delay(4) {
_ = intSequence
.subscribe(onNext: { print("Subscription 2:, Event: \($0)") })
}
delay(8) {
_ = intSequence
.subscribe(onNext: { print("Subscription 3:, Event: \($0)") })
}
Subscription 1:, Event: 0
Subscription 1:, Event: 1
Subscription 1:, Event: 2
Subscription 2:, Event: 0
Subscription 2:, Event: 1
Subscription 2:, Event: 2
Subscription 1:, Event: 3
Subscription 2:, Event: 3
Subscription 1:, Event: 4
Subscription 2:, Event: 4
Subscription 1:, Event: 5
Subscription 2:, Event: 5
Subscription 1:, Event: 6
Subscription 2:, Event: 6
Subscription 3:, Event: 2
Subscription 3:, Event: 3
Subscription 3:, Event: 4
Subscription 3:, Event: 5
Subscription 3:, Event: 6
Subscription 1:, Event: 7
Subscription 2:, Event: 7
Subscription 3:, Event: 7
multicast:
Converts the source Observable
sequence into a connectable sequence, and broadcasts its emissions via the specified subject
.
将源“Observable”序列转换为可连接的序列,并通过指定的“subject”广播其发射。会先发射“subject”的事件。
let subject = PublishSubject<Int>()
_ = subject
.subscribe(onNext: { print("Subject: \($0)") })
let intSequence = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
.multicast(subject)
_ = intSequence
.subscribe(onNext: { print("\tSubscription 1:, Event: \($0)") })
delay(2) { _ = intSequence.connect() }
delay(4) {
_ = intSequence
.subscribe(onNext: { print("\tSubscription 2:, Event: \($0)") })
}
delay(6) {
_ = intSequence
.subscribe(onNext: { print("\tSubscription 3:, Event: \($0)") })
}
Subject: 0
Subscription 1:, Event: 0
Subject: 1
Subscription 1:, Event: 1
Subscription 2:, Event: 1
Subject: 2
Subscription 1:, Event: 2
Subscription 2:, Event: 2
Subject: 3
Subscription 1:, Event: 3
Subscription 2:, Event: 3
Subscription 3:, Event: 3
Subject: 4
Subscription 1:, Event: 4
Subscription 2:, Event: 4
Subscription 3:, Event: 4
Subject: 5
Subscription 1:, Event: 5
Subscription 2:, Event: 5
Subscription 3:, Event: 5
Error Handling Operators:
Operators that help to recover from error notifications from an Observable.
有助于从Observable的错误通知中恢复的操作符。
catchErrorJustReturn:
Recovers from an Error event by returning an Observable
sequence that emits a single element and then terminates. More info
通过返回发出单个元素的Observable
序列,从一个原始序列中恢复错误,然后终止。
[图片上传失败...(image-29b9cd-1515048671654)]
let disposeBag = DisposeBag()
let sequenceThatFails = PublishSubject<String>()
sequenceThatFails
.catchErrorJustReturn("😊")
.subscribe { print($0) }
.disposed(by: disposeBag)
sequenceThatFails.onNext("😬")
sequenceThatFails.onNext("😨")
sequenceThatFails.onNext("😡")
sequenceThatFails.onNext("🔴")
sequenceThatFails.onError(TestError.test)
next(😬)
next(😨)
next(😡)
next(🔴)
next(😊)
completed
catchError:
Recovers from an Error event by switching to the provided recovery Observable
sequence. More info
通过转换到提供的恢复Observable
序列,恢复一个错误。
[图片上传失败...(image-e72e5c-1515048671654)]
let disposeBag = DisposeBag()
let sequenceThatFails = PublishSubject<String>()
let recoverySequence = PublishSubject<String>()
sequenceThatFails
.catchError {
print("Error:", $0)
return recoverySequence
}
.subscribe { print($0) }
.disposed(by: disposeBag)
sequenceThatFails.onNext("😬")
sequenceThatFails.onNext("😨")
sequenceThatFails.onNext("😡")
sequenceThatFails.onNext("🔴")
sequenceThatFails.onError(TestError.test)
recoverySequence.onNext("😊")
next(😬)
next(😨)
next(😡)
next(🔴)
Error: test
next(😊)
retry:
Recovers repeatedly Error events by resubscribing to the Observable
sequence, indefinitely. More info
通过无限期地重新订阅“可观察”序列来重复恢复错误事件。
[图片上传失败...(image-46b3bf-1515048671654)]
let disposeBag = DisposeBag()
var count = 1
let sequenceThatErrors = Observable<String>.create { observer in
observer.onNext("🍎")
observer.onNext("🍐")
observer.onNext("🍊")
if count == 1 {
observer.onError(TestError.test)
print("Error encountered")
count += 1
}
observer.onNext("🐶")
observer.onNext("🐱")
observer.onNext("🐭")
observer.onCompleted()
return Disposables.create()
}
sequenceThatErrors
.retry()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🍎
🍐
🍊
Error encountered
🍎
🍐
🍊
🐶
🐱
🐭
retry(_:):
Recovers repeatedly from Error events by resubscribing to the Observable
sequence, up to maxAttemptCount
number of retries. More info
通过重新订阅“Observable”序列,直到“maxAttemptCount”重试次数,从错误事件中重复恢复
[图片上传失败...(image-9cffd1-1515048671654)]
let disposeBag = DisposeBag()
var count = 1
let sequenceThatErrors = Observable<String>.create { observer in
observer.onNext("🍎")
observer.onNext("🍐")
observer.onNext("🍊")
if count < 5 {
observer.onError(TestError.test)
print("Error encountered")
count += 1
}
observer.onNext("🐶")
observer.onNext("🐱")
observer.onNext("🐭")
observer.onCompleted()
return Disposables.create()
}
sequenceThatErrors
.retry(3)
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
🍎
🍐
🍊
Error encountered
🍎
🍐
🍊
Error encountered
🍎
🍐
🍊
Error encountered
Debugging Operators:Operators to help debug Rx code.
帮助调试Rx代码的操作符。
debug:
Prints out all subscriptions, events, and disposals.
打印出所有的订阅,事件和处理。
let disposeBag = DisposeBag()
var count = 1
let sequenceThatErrors = Observable<String>.create { observer in
observer.onNext("🍎")
observer.onNext("🍐")
observer.onNext("🍊")
if count < 5 {
observer.onError(TestError.test)
print("Error encountered")
count += 1
}
observer.onNext("🐶")
observer.onNext("🐱")
observer.onNext("🐭")
observer.onCompleted()
return Disposables.create()
}
sequenceThatErrors
.retry(3)
.debug()
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
2017-12-08 09:43:16.702: Rx.playground:42 (__lldb_expr_289) -> subscribed
2017-12-08 09:43:16.704: Rx.playground:42 (__lldb_expr_289) -> Event next(🍎)
🍎
2017-12-08 09:43:16.704: Rx.playground:42 (__lldb_expr_289) -> Event next(🍐)
🍐
2017-12-08 09:43:16.704: Rx.playground:42 (__lldb_expr_289) -> Event next(🍊)
🍊
Error encountered
2017-12-08 09:43:16.706: Rx.playground:42 (__lldb_expr_289) -> Event next(🍎)
🍎
2017-12-08 09:43:16.706: Rx.playground:42 (__lldb_expr_289) -> Event next(🍐)
🍐
2017-12-08 09:43:16.706: Rx.playground:42 (__lldb_expr_289) -> Event next(🍊)
🍊
Error encountered
2017-12-08 09:43:16.708: Rx.playground:42 (__lldb_expr_289) -> Event next(🍎)
🍎
2017-12-08 09:43:16.708: Rx.playground:42 (__lldb_expr_289) -> Event next(🍐)
🍐
2017-12-08 09:43:16.708: Rx.playground:42 (__lldb_expr_289) -> Event next(🍊)
🍊
Error encountered
2017-12-08 09:43:16.709: Rx.playground:42 (__lldb_expr_289) -> Event error(test)
Unhandled error happened: test
RxSwift.Resources.total:
Provides a count of all Rx resource allocations, which is useful for detecting leaks during development.
提供所有Rx资源分配的计数,这对于在开发过程中检测泄漏非常有用。
example("RxSwift.Resources.total") {
print(RxSwift.Resources.total)
let disposeBag = DisposeBag()
print(RxSwift.Resources.total)
let variable = Variable("🍎")
let subscription1 = variable.asObservable().subscribe(onNext: { print($0) })
print(RxSwift.Resources.total)
let subscription2 = variable.asObservable().subscribe(onNext: { print($0) })
print(RxSwift.Resources.total)
subscription1.dispose()
print(RxSwift.Resources.total)
subscription2.dispose()
print(RxSwift.Resources.total)
}
print(RxSwift.Resources.total)
1
3
🍎
12
🍎
15
13
11
1