#はじめに
前回
今回はOperatorをいくつか練習していきます

#Operators

##Filtering Operators

###ignoreElements
onNextを無視する

let publishSubject = PublishSubject<String>()
let disposeBag = DisposeBag()
publishSubject
    .ignoreElements()
    .subscribe { _ in
        print("subscription is called")
    }
    .disposed(by: disposeBag)

publishSubject.onNext("1") // 呼ばれない
publishSubject.onNext("2") // 呼ばれない
publishSubject.onNext("3") // 呼ばれない
publishSubject.onCompleted() // "subscription is called"

###element(at:)
指定した箇所のイベントを購読

let publishSubject = PublishSubject<String>()
publishSubject
    .element(at: 1)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)

publishSubject.onNext("1")
publishSubject.onNext("2") // 2
publishSubject.onNext("3")

###filter
条件に合うものだけを購読

Observable.of("aa", "bbb", "cccc", "ddddd", "eeeeee")
    .filter { $0.count > 3 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

//cccc
//ddddd
//eeeeee

###skip
指定した要素の数だけイベントを飛ばす

Observable.of(1, 2, 3, 4, 5, 6)
    .skip(3)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

// 4
// 5
// 6

###skip(while:)
条件に合うイベントまでスキップする

Observable.of(1, 1, 1, 2, 2, 2, 3, 3, 3)
    .skip(while: { $0 != 2 }) // 2じゃない間はとばし、2になったらそれ以降購読
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

// 2
// 2
// 2
// 3
// 3
// 3

###skip(until: )
基準のイベントが流れてきたらそれ以降を購読

let trigger = PublishSubject<String>()
publishSubject.skip(until: trigger)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

publishSubject.onNext("1")
publishSubject.onNext("2")
trigger.onNext("X")
publishSubject.onNext("3")
publishSubject.onNext("4")

// 3
// 4

###take
引数の数だけ、先頭から購読

Observable.of(1, 2, 3, 4, 5, 6)
    .take(3)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

// 1
// 2
// 3

###take(while: )
条件に合う限り購読、一度でも条件を満た差なくなるとそれ以降購読しない

Observable.of(1, 2, 3, 4, 5, 6)
    .take(while: { $0 < 4 } )
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

// 1
// 2
// 3

###take(until: )
引数に基準となるオブザーバブルを取り、それが購読されるまでのイベントを購読する

let trigger = PublishSubject<String>()
publishSubject
    .take(until: trigger)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

publishSubject.onNext("1")
publishSubject.onNext("2")
trigger.onNext("X")
publishSubject.onNext("3")
publishSubject.onNext("4")

// 1
// 2

##Transforming Operators

###toArray
配列にする

Observable.of(1, 2, 3, 4, 5)
    .toArray()
    .subscribe(onSuccess: { print($0) },
               onFailure: { print($0) },
               onDisposed: { print("onDisposed") }
    ).disposed(by: disposeBag)

// [1, 2, 3, 4, 5]
// onDisposed

###map
全てのイベントを変換

Observable.of(1, 2, 3, 4, 5)
    .map { $0 * 10 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

//10
//20
//30
//40
//50

###flatMap
map + mergeのようなイメージ

struct Student {
    var score: BehaviorRelay<Int>
}
let reon = Student(score: BehaviorRelay(value: 10))
let tom = Student(score: BehaviorRelay(value: 20))
let student = PublishSubject<Student>()
student.asObservable()
    .flatMap { $0.score.asObservable() }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

student.onNext(reon) // 10
reon.score.accept(100) // 100
tom.score.accept(200) // 200
student.onNext(tom) // 呼ばれない

###flatMapLatest
map + switchLatest(次のObservableが来たらそちらを優先する)

struct Student {
    var score: BehaviorRelay<Int>
}
let reon = Student(score: BehaviorRelay(value: 10))
let tom = Student(score: BehaviorRelay(value: 20))
let student = PublishSubject<Student>()
student.asObservable()
    .flatMapLatest { $0.score.asObservable() }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

student.onNext(reon)
student.onNext(tom)
reon.score.accept(100)
tom.score.accept(200)

##Combing Operators

###startsWith
先頭のイベントを任意のものに変更できる

let numbers = Observable.of(2, 3, 4)
let observable = numbers.startWith(1)
observable
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// 1
// 2
// 3
// 4

###concat
Observableを結合できる

let first = Observable.of(1, 2, 3)
let second = Observable.of(4, 5, 6)
let observable = Observable.concat([first, second])
observable
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// 1
// 2
// 3
// 4
// 5
// 6

###merge
こちらもObservableを結合できますが、concatとの違いは、時系列順になるということです。

let left = PublishSubject<Int>()
let right = PublishSubject<Int>()
let source = Observable.of(left.asObservable(), right.asObservable())
let observable = source.merge()
observable
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

left.onNext(1)
left.onNext(2)
right.onNext(100)
left.onNext(3)
right.onNext(200)
right.onNext(300)

//1
//2
//100
//3
//200
//300

###combineLatest
近い最新の値同士を組み合わせたイベントを作成

let left = PublishSubject<String>()
let right = PublishSubject<String>()
let observable = Observable.combineLatest(left, right) { lastLeft, lastRight in
    return lastLeft + lastRight
}
let disposable = observable.subscribe(onNext: { print($0) } )
left.onNext("A")
left.onNext("B")
right.onNext("1")
left.onNext("C")
right.onNext("2")
right.onNext("3")
left.onNext("D")
right.onNext("4")

// B1
// C1
// C2
// C3
// D3
// D4

###withLatestFrom
任意のイベント発生時に、最新のイベントを発生させる。
今回は、ボタンを押した後に、テキストフィールドの最新のイベント(Swift)を流している。

let button = PublishSubject<Void>()
let textField = PublishSubject<String>()
let observable = button.withLatestFrom(textField)
let disposable = observable.subscribe(onNext: { print($0) })

textField.onNext("S")
textField.onNext("Sw")
textField.onNext("Swi")
textField.onNext("Swif")
textField.onNext("Swift") // これがtextFieldの最新の値
button.onNext(()) // ここで初めて呼ばれる
button.onNext(()) // ここでも呼ばれる

// Swift
// Swift

###reduce
イベントを一つにまとめる加工を行う

let observable = Observable.of(1, 2, 3)
observable.reduce(0, accumulator: +)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

// 6

###scan
reduceの時と違い、日々イベントを更新していく

let observable = Observable.of(1, 2, 3)
observable.scan(0, accumulator: +)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// 1
// 3
// 6

#おわりに
次回
他にもたくさん種類があるので、気になる方はマーブルダイアグラムを見てみてください！
マーブルダイアグラム