Posted at

【読書メモ】RxSwift: Reactive Programming with Swift, Chapter 5~11

More than 1 year has passed since last update.



  • RxSwift: Reactive Programming with Swift の読書メモです。この本を読んでみようかと思う人の参考になれば。

  • 演習が充実している本ですが、その部分については割愛します。


Chapter 5: Filtering Operators


Ignoring operators

// protocol ObservableType

func ignoreElements() -> RxSwift.Observable<Self.E>

// protocol ObservableType

func elementAt(_ index: Int) -> RxSwift.Observable<Self.E>

// protocol ObservableType

func filter(_ predicate: @escaping (Self.E) throws -> Bool) -> RxSwift.Observable<Self.E>


Skipping operators

// protocol ObservableType

func skip(_ count: Int) -> RxSwift.Observable<Self.E>


  • skipWhile


    • クロージャで指定した条件が true の間要素を無視する。一度 false になったらそれ以降の要素を全て通過させる。

    • クロージャに要素とそのインデクス番号が渡ってくる skipWhileWithIndex もある。


    • http://reactivex.io/documentation/operators/skipwhile.html




// protocol ObservableType

func skipWhile(_ predicate: @escaping (Self.E) throws -> Bool) -> RxSwift.Observable<Self.E>


Taking operators

// protocol ObservableType

func take(_ count: Int) -> RxSwift.Observable<Self.E>


  • takeWhile


    • クロージャで指定した条件が true の間要素を通過させる。一度 false になったら以降の要素を全て無視する。

    • クロージャに要素とそのインデックス番号が渡ってくる takeWhileWithIndex もある。

    • 本の takeWhilewithIndex のマーブル図はおかしい。


    • http://reactivex.io/documentation/operators/takewhile.html




// protocol ObservableType

func takeWhile(_ predicate: @escaping (Self.E) throws -> Bool) -> RxSwift.Observable<Self.E>

// protocol ObservableType

func takeUntil<O>(_ other: O) -> RxSwift.Observable<Self.E> where O : ObservableType


Distinct operators

// protocol ObservableType where Self.E : Equatable

func distinctUntilChanged() -> RxSwift.Observable<Self.E>
// protocol ObservableType
func distinctUntilChanged(_ comparer: @escaping (Self.E, Self.E) throws -> Bool) -> RxSwift.Observable<Self.E>


chapter 6: Filtering Operators in Practice


Sharing subscriptions


  • observable にオペレータを連ねただでけは実際には何もしない。末尾で subscribe() を呼んで初めて値の生成を行う。

  • 通常 subscribe() を呼ぶ毎に新しい Observable が作られるが、それらが同じもので同じ値を返す保証はない。また同じ値を返すとしても生成コストが大きい場合がある。この場合 share() で作られる共有 Observable が使える。共有 Observable は参照カウンタを持っていて最初に subscribe される時に作られ全てのサブスクリプションから dispose() される時に破棄される。その後再度 subscribe される時には新しいインスタンスが作られる。

// protocol ObservableType

func share() -> RxSwift.Observable<Self.E>


  • 指定した時間の間イベントを通過させ、そのあとに completed を送る。

// protocol ObservableType

func take(_ duration: RxTimeInterval, scheduler: SchedulerType) -> RxSwift.Observable<Self.E>


  • throttle で一定期間入力を遅延させ、その間の最後の入力のみを出力させることができる。これによって短期間に 2 回タップしたを 1 回として扱ったり、インクリメンタルサーチできたりする。


  • http://reactivex.io/documentation/operators/debounce.html


// protocol ObservableType

func throttle(_ dueTime: RxTimeInterval, latest: Bool = default, scheduler: SchedulerType) -> RxSwift.Observable<Self.E>


Chapter 7: Transforming Operators


Transforming elements

// protocol ObservableType

func toArray() -> RxSwift.Observable<[Self.E]>

// protocol ObservableType

func map<R>(_ transform: @escaping (Self.E) throws -> R) -> RxSwift.Observable<R>
func mapWithIndex<R>(_ selector: @escaping (Self.E, Int) throws -> R) -> RxSwift.Observable<R>


Transforming inner observables


  • flatMap


    • Observable (または Observable に変換可能な要素)のシーケンスから各 Observable の要素が全て流れてくるシーケンスを作る。

    • 複数のシーケンスを一つのシーケンスにまとめる。

    • 引数に渡すクロージャで Observable に変換可能な要素を Observable に変換する。もともと Observable ならそのまま返す。

    • クロージャは各 Observable にごとに 1 回呼ばれる。各 Observalbe に値が流れて来るたびに呼ばれるわけではない。


    • http://reactivex.io/documentation/operators/flatmap.html




// protocol ObservableType

func flatMap<O>(_ selector: @escaping (Self.E) throws -> O) -> RxSwift.Observable<O.E> where O : ObservableConvertibleType


  • flatMapLatest


    • 元のシーケンス (Observable のシーケンス) に追加された最後の Observable から流れて来る要素だけを結果のシーケンスに流す。

    • 一例としてはインテリセンスみたいなものを実装する際に使える。最新の入力仕掛けの文字列に対するサジェスチョンのみを表示したい場合、過去の入力仕掛けの文字列に対して非同期で返ってきたサジェスチョンを無視するのに使える。


    • http://reactivex.io/documentation/operators/flatmap.html




// protocol ObservableType

func flatMapLatest<O>(_ selector: @escaping (Self.E) throws -> O) -> RxSwift.Observable<O.E> where O : ObservableConvertibleType


Chapter 8: Transforming Operators in Practice


  • RxSwift はプラットフォームに依存しない純粋な Rx API を提供するのに対し、RxCocoa は UIKit と Cocoa に固有な機能を提供する。

  • shareReply



// protocol ObsevableType

func shareReplay(_ bufferSize: Int) -> RxSwift.Observable<Self.E>


Chapter 9: Combining Operators


Prefixing and concatenating

// protocol ObservableType

func startWith(_ elements: Self.E...) -> RxSwift.Observable<Self.E>


  • concat


    • 完了する複数のシーケンスを繋げる。まず最初のシーケンスの要素が出力され、完了したあとに次のシーケンスの要素が出力され、その完了後に 3 番目のシーケンスが...と続く。

    • いずれかのシーケンスがエラーになると concat の結果のシーケンスもエラーになる。


    • http://reactivex.io/documentation/operators/concat.html




// static method

// class Observable
static func concat<S: Sequence >(_ sequence: S) -> Observable<Element> where S.Iterator.Element == Observable<Element>
static func concat<S: Collection >(_ collection: S) -> Observable<Element> where S.Iterator.Element == Observable<Element>
static func concat(_ sources: Observable<Element> ...) -> Observable<Element>

// instance method
// protocol ObservableType
func concat<O>(_ second: O) -> RxSwift.Observable<Self.E> where O : ObservableConvertibleType, O.E == Self.E


Merging


  • merge


    • 複数のシーケンスから単一のシーケンスを作る。入力シーケンスのいずれかの値が変化した時にその値が出力シーケンスの値として出力される。

    • 元になった全てのシーケンスが終了するとそのシーケンスも終了する。

    • 元になったシーケンスのいずれかが終了するとそのシーケンスも終了する。

    • 同時にサブスクライブするシーケンスの最大数を制限できる変種がある。


    • http://reactivex.io/documentation/operators/merge.html




// protocol ObservableType where Self.E : ObservableConvertibleType

func merge() -> RxSwift.Observable<Self.E.E>
func merge(maxConcurrent: Int) -> RxSwift.Observable<Self.E.E>


Combining elements


  • combineLatest


    • 複数のシーケンスから単一のシーケンスを作る。入力シーケンスのいずれかの値が変化した時に、各入力シーケンスの最新の値を引数に取るクロージャが実行されその結果が出力シーケンスの値となる。

    • 全ての入力シーケンスから最初の値が入って来るまで出力シーケンスに値は出力されない。

    • 各入力シーケンスの値が独立した引数としてクロージャに渡って来る変種とそれらを一つの配列に詰め込んで渡って来る変種がある。


    • http://reactivex.io/documentation/operators/combinelatest.html




// class Observable

static func combineLatest<O1: ObservableType, O2: ObservableType>(_ source1: O1, _ source2: O2, resultSelector: @escaping (O1.E, O2.E) throws -> E) -> Observable<E>
static func combineLatest<C: Collection>(_ collection: C, _ resultSelector: @escaping ([C.Iterator.Element.E]) throws -> Element) -> Observable<Element> where C.Iterator.Element: ObservableType


  • zip


    • 複数の入力シーケンスから単一のシーケンスを作る。入力シーケンスの全ての値が変化した時に、各入力シーケンスの値を引数に取るクロージャが実行されその結果が出力シーケンスの値となる。

    • 全ての入力シーケンスの値が変化するまで出力シーケンスに新しい値は出力されない。


    • http://reactivex.io/documentation/operators/zip.html




// class Observable

static func zip<O1: ObservableType, O2: ObservableType>(_ source1: O1, _ source2: O2, resultSelector: @escaping (O1.E, O2.E) throws -> E) -> Observable<E>


Triggers

// extension ObservableType

func withLatestFrom<SecondO>(_ second: SecondO) -> RxSwift.Observable<SecondO.E> where SecondO : ObservableConvertibleType


  • sample


    • trigger の変種でトリガーとなるデータソースにイベントが発生し、かつ、データソースのデータが前回と変わっていた場合だデータを出力する。

    • trigger とはトリガーとデータソースの指定が逆になっているので注意。


    • http://reactivex.io/documentation/operators/sample.html




// protocol ObservableType

func sample<O>(_ sampler: O) -> RxSwift.Observable<Self.E> where O : ObservableType


Switches

// protocol ObservableType

func amb<O2>(_ right: O2) -> RxSwift.Observable<Self.E> where O2 : ObservableType, O2.E == Self.E

// protocol ObservableType where Self.E : ObservableConvertibleType

func switchLatest() -> RxSwift.Observable<Self.E.E>


Combining elements within a sequence

// protocol ObservableType

func reduce<A>(_ seed: A, accumulator: @escaping (A, Self.E) throws -> A) -> RxSwift.Observable<A>


  • scan


    • reduce とほぼ同じ。

    • 新しい値が入力され、再計算される度に値が出力される点が reduce と異なる。

    • reduce, scan はステートを持つので上手に使うと変数を使わずに済ませられるかも


    • http://reactivex.io/documentation/operators/scan.html




// protocol ObservableType

func scan<A>(_ seed: A, accumulator: @escaping (A, Self.E) throws -> A) -> RxSwift.Observable<A>


Chapter 10: Combining Operators in Practice

(演習なので割愛)


Chapter 11: Time Based Operators


Buffering operators


  • replay


    • 指定の数の最新の要素を保持しておき、新しく接続してきた Observer にはまずそれを渡す。それ以後は入力から渡って来る要素をそのまま渡す。

    • replyAll は全ての要素を保持する。

    • 戻り値である ConnectableObservable は subscribe するだけではイベントを発行せず、connect() が呼ばれて初めて発行する。


    • http://reactivex.io/documentation/operators/replay.html




// protocol ObservableType

func replay(_ bufferSize: Int) -> RxSwift.ConnectableObservable<Self.E>


  • buffer


    • 入力データのバッファリングを行う。


      • バッファは指定されたサイズの要素を保持することができる。

      • 入力されたデータは一旦バッファに保存する。

      • バッファリング時間の間新たな要素の入力がなかったらバッファに溜まっているデータを全て配列として出力する。

      • バッファに保持できる以上のデータの入力あった場合は即座にバッファの中の最も古いデータを出力し、新しいデータをバッファに保持する。

      • 新たなデータがなかった場合でもバッファリング時間毎に要素 0 個の配列を出力する。




    • http://reactivex.io/documentation/operators/buffer.html




// protocol Observabletype

func buffer(timeSpan: RxTimeInterval, count: Int, scheduler: SchedulerType) -> Observable<[E]>


  • window


    • buffer とほぼ同じだが、buffer は一定周期またはバッファが一杯になる毎に要素の配列を返すのに対し、windows は要素を出力する Observable を返す。

    • 一定周期またバッファが一杯になる度に出力先の Observable を変更する。


    • http://reactivex.io/documentation/operators/window.html




// protocol ObservableType

func window(timeSpan: RxTimeInterval, count: Int, scheduler: SchedulerType) -> RxSwift.Observable<RxSwift.Observable<Self.E>>


Time-shifting operators

// protocol ObservableType

func delay(_ dueTime: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>


  • delaySubscription


    • サブスクリプションの開始を遅延させる。

    • delay は全ての要素の出力を遅延させるのに対し delaySubscription はサブスクリプションの開始のみを遅延させる。サブスクリプション開始前に入力された値は失われる。

    • サブスクリプション開始後は入力が遅延なしで出力させる。


    • http://reactivex.io/documentation/operators/delay.html




// protocol ObservableType

func delaySubscription(_ dueTime: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>


Time Operators

// class Observable where Element : SignedInteger

static func interval(_ period: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>


  • timer


    • 一定時間毎に値を出力する Observable を作る。

    • 値は符号付き整数で 0 からインクリメントしていく。

    • dueTime は最初のイベント発行までの時間。

    • period はその後のイベントの発行間隔。nil だとワンショットになる。


    • http://reactivex.io/documentation/operators/timer.html




// class Observable where Element: SignedInteger

static func timer(_ dueTime: RxTimeInterval, period: RxTimeInterval? = nil, scheduler: SchedulerType) -> Observable<E>


  • timeout


    • 一定時間入力がなかった場合にエラーイベント(RxError.timeoutError)を発行する。

    • エラーイベントを発行するのではなく、入力元を別の Observable に切り替える変種もある。


    • http://reactivex.io/documentation/operators/timeout.html




// protocol ObservableType

func timeout(_ dueTime: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>
func timeout<O: ObservableConvertibleType>(_ dueTime: RxTimeInterval, other: O, scheduler: SchedulerType) -> Observable<E> where E == O.E