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
一個(gè)Subject是一種橋梁或代理，在Rx的一些實(shí)現(xiàn)中可用作觀察者和Observable槽华。因?yàn)樗且粋€(gè)觀察者涣达，所以可以訂閱一個(gè)或多個(gè)Observable，由于它是一個(gè)Observable厦坛，它可以通過(guò)重新發(fā)送它們來(lái)通過(guò)它觀察到的項(xiàng)目够傍，也可以發(fā)出新的項(xiàng)目。

    /**
     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.
在觀察者訂閱的時(shí)間向它們廣播新的事件。

    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.
向所有訂閱者廣播新事件蚓挤，并向新訂閱者廣播指定以前事件序列中緩沖區(qū)大小數(shù)量的事件（從后往前數(shù)）。

    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.

向所有用戶廣播新的事件驻子，并向新的用戶提供最近（或最初）的價(jià)值灿意。

    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.
包裝一個(gè)BehaviorSubject，所以它會(huì)發(fā)送最新的（或初始）值給新的觀察者拴孤。Variable也保持當(dāng)前值狀態(tài)脾歧。Variable將永遠(yuǎn)不會(huì)發(fā)出一個(gè)錯(cuò)誤事件。但是演熟，它會(huì)自動(dòng)發(fā)出一個(gè)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

Subject詳細(xì)介紹

Combining_Operators 聯(lián)合操作符

聯(lián)合多個(gè)Observable到一個(gè)Observable

startWith:

Emits the specified sequence of elements before beginning to emit the elements from the source Observable. More info
在開(kāi)始發(fā)送原始的元素序列之前芒粹，發(fā)射指定的元素序列兄纺。
你可以鏈?zhǔn)降膶?duì)一個(gè)Observable調(diào)用多次startWith，但是發(fā)射的序列是按后進(jìn)先出（last-in-first-out）的方式化漆，即后startWith的元素序列會(huì)先發(fā)射出去

    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
結(jié)合多個(gè)發(fā)射序列為一個(gè)新的序列

    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個(gè)源“Observable”序列組合成一個(gè)新的“Observable”序列估脆，并且將從組合的“Observable”序列發(fā)出來(lái)自相應(yīng)索引處的每個(gè)源“Observable”序列的元素。發(fā)射的序列個(gè)數(shù)是根據(jù)最少的序列個(gè)數(shù)來(lái)定的

    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個(gè)源Observable序列組合成一個(gè)新的Observable序列座云，并且一旦所有源序列發(fā)射至少一個(gè)元素疙赠，將從組合的Observable序列開(kāi)始發(fā)射每個(gè)源“可觀察”序列的最新元素，還有當(dāng)任何源Observable序列發(fā)出一個(gè)新的元素時(shí)候朦拖，也從其它Observable中拿出最近的元素來(lái)組合發(fā)射

    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
將由觀察序列發(fā)出的元素轉(zhuǎn)換為觀察序列圃阳，并且只有最新的觀察序列能發(fā)送信號(hào)

    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
    //此時(shí)不會(huì)打印??因?yàn)楫?dāng)前最新的variable是subject2
    subject1.onNext("??")
    
    subject2.onNext("??")

??
??
??
??
??

Transforming_Operators:

Operators that transform Next event elements emitted by an Observable sequence.
轉(zhuǎn)換操作符：轉(zhuǎn)換發(fā)射的Next事件元素到一個(gè)新的Observable序列

map:

Applies a transforming closure to elements emitted by an Observable sequence, and returns a new Observable sequence of the transformed elements.
將變換閉包應(yīng)用于由“Observable”序列發(fā)出的元素，并返回變換元素的一個(gè)新的“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.
將由“可觀察”序列發(fā)射的元素轉(zhuǎn)換為“可觀察”序列捍岳，并將來(lái)自兩個(gè)“可觀察”序列的發(fā)射合并成單個(gè)“可觀察”序列。例如睬隶，當(dāng)你有一個(gè)本身發(fā)出“可觀察”序列的“可觀察”序列锣夹，并且你希望能夠?qū)?lái)自“可觀察”序列的新發(fā)射作出反應(yīng)時(shí)，這也是有用的苏潜。 “flatMap”和“flatMapLatest”的區(qū)別在于银萍，“flatMapLatest”只會(huì)從最近的“Observable”內(nèi)部發(fā)出元素。
flatMapLatest是map和switchLatest的結(jié)合體恤左。

    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.
從一個(gè)初始的種子值開(kāi)始贴唇，然后對(duì)由“可觀察”序列發(fā)出的每個(gè)元素應(yīng)用累加器閉包贰锁，并將每個(gè)中間結(jié)果作為單元素“可觀察”序列返回。

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”序列中發(fā)射元素的運(yùn)算符

filter:

Emits only those elements from an Observable sequence that meet the specified condition.

僅從符合指定條件的“可觀察”序列中發(fā)射那些元素

    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
抑制由“可觀察”序列發(fā)出的連續(xù)重復(fù)元素滤蝠。
[圖片上傳失敗...(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
僅發(fā)出由“Observable”序列發(fā)出的所有元素的指定索引處的元素
[圖片上傳失敗...(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.
僅發(fā)出由“可觀察”序列發(fā)出的第一個(gè)元素（或符合條件的第一個(gè)元素）。如果Observable序列不是只發(fā)出一個(gè)元素授嘀，會(huì)拋出一個(gè)錯(cuò)誤物咳。

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
僅從“可觀察”序列的開(kāi)始處發(fā)射指定數(shù)量的元素。

[圖片上傳失敗...(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
僅從"可觀察"序列的末尾處發(fā)射指定數(shù)量的元素蹄皱。注意：發(fā)射元素的順序不變览闰，只不過(guò)是從后往前數(shù)，從第幾個(gè)開(kāi)始發(fā)射巷折。
[圖片上傳失敗...(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
只要指定的條件評(píng)估為“真”压鉴，就從“可觀察”序列的開(kāi)始處發(fā)射元素。從開(kāi)始往后锻拘，如果有元素不滿足條件油吭，將發(fā)射completed事件而不再發(fā)射元素。

[圖片上傳失敗...(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序列中發(fā)射元素署拟，直到參考Observable序列發(fā)射一個(gè)元素婉宰。

[圖片上傳失敗...(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
禁止從“可觀察”序列的開(kāi)始發(fā)射指定數(shù)量的元素。

[圖片上傳失敗...(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
禁止從符合指定條件的“可觀察”序列開(kāi)始發(fā)射元素推穷。

[圖片上傳失敗...(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.
禁止從符合指定條件的“可觀察”序列開(kāi)始發(fā)射元素心包，并發(fā)射其余元素。閉包也傳遞每個(gè)元素的索引馒铃。

    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”序列發(fā)射元素蟹腾，直到引用“Observable”序列發(fā)射一個(gè)元素馍管。

[圖片上傳失敗...(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.

對(duì)Observable發(fā)出的所有項(xiàng)目進(jìn)行操作的操作符奠涌。

toArray:

Converts an Observable sequence into an array, emits that array as a new single-element Observable sequence, and then terminates. More info

將一個(gè)Observable序列轉(zhuǎn)換為一個(gè)數(shù)組，將該數(shù)組作為一個(gè)新的單元素Observable序列發(fā)送世蔗，然后終止萧锉。

[圖片上傳失敗...(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
以一個(gè)初始的種子值開(kāi)始珊随，然后將一個(gè)累加器閉包應(yīng)用于由一個(gè)“可觀察”序列發(fā)出的所有元素，并且將該聚合結(jié)果作為單元素“可觀察”序列返回柿隙。

[圖片上傳失敗...(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
以順序的方式連接來(lái)自“Observable”序列的內(nèi)部“可觀察”序列的元素叶洞，等待每個(gè)序列在從下一個(gè)序列發(fā)射元素之前成功終止。

[圖片上傳失敗...(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序列相似禀崖，只不過(guò)它們?cè)谟嗛啎r(shí)不會(huì)開(kāi)始發(fā)射元素衩辟，而只是在調(diào)用connect（）方法的時(shí)候。通過(guò)這種方式波附，您可以等待所有預(yù)定用戶在開(kāi)始發(fā)射元素之前預(yù)訂可連接的“可觀察”序列艺晴。

interval:

interval creates an Observable sequence that emits elements after each period, on the specified scheduler. More info
interval在指定的調(diào)度器上創(chuàng)建一個(gè)Observable序列昼钻，在每個(gè)`period“之后發(fā)出元素。
[圖片上傳失敗...(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”序列轉(zhuǎn)換為可連接的序列封寞。

[圖片上傳失敗...(image-70b1a9-1515048671654)]

    let intSequence = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        .publish()
    
    _ = intSequence
        .subscribe(onNext: { print("Subscription 1:, Event: \($0)") })
        
    //如果我把這個(gè)connect延遲設(shè)為6的話然评，3個(gè)觀察會(huì)一起打印
    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”序列轉(zhuǎn)換為可連接的序列，并將向每個(gè)新訂戶播放先前發(fā)射的“bufferSize”數(shù)量狈究。
[圖片上傳失敗...(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”序列轉(zhuǎn)換為可連接的序列碗淌，并通過(guò)指定的“subject”廣播其發(fā)射。會(huì)先發(fā)射“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的錯(cuò)誤通知中恢復(fù)的操作符亿眠。

catchErrorJustReturn：

Recovers from an Error event by returning an Observable sequence that emits a single element and then terminates. More info
通過(guò)返回發(fā)出單個(gè)元素的Observable序列，從一個(gè)原始序列中恢復(fù)錯(cuò)誤磅废，然后終止纳像。
[圖片上傳失敗...(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
通過(guò)轉(zhuǎn)換到提供的恢復(fù)Observable序列，恢復(fù)一個(gè)錯(cuò)誤拯勉。

[圖片上傳失敗...(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
通過(guò)無(wú)限期地重新訂閱“可觀察”序列來(lái)重復(fù)恢復(fù)錯(cuò)誤事件竟趾。
[圖片上傳失敗...(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
通過(guò)重新訂閱“Observable”序列，直到“maxAttemptCount”重試次數(shù)谜喊，從錯(cuò)誤事件中重復(fù)恢復(fù)
[圖片上傳失敗...(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.

幫助調(diào)試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資源分配的計(jì)數(shù)山卦，這對(duì)于在開(kāi)發(fā)過(guò)程中檢測(cè)泄漏非常有用。

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