RxJavaの基本的な処理シーケンス（古め）

Reactiveの勉強記事

Reactiveを理解したいが何もわからんので、
Reactive Programming With RxJava のサンプルコードが、裏でどのように動いているのかを記述していく。

2016年の本なので、Observable.create のような Deprecatedとなっている古いAPIなどを使用している。

RxJava

簡単な登場人物

シンプル

fun main(args: Array<String>) {
    Observable.create<String> { s ->
        s.onNext("one")
        s.onCompleted()
    }.subscribe { s -> println("VALUE -> $s") }
}

• create {s -> s.OnNext("one") ...}
  • 引数のラムダ式は OnSubscribe の実装として解釈される。
• subscribe { s -> println("VALUE -> $s") }
  • 引数のラムダ式は Action1 という内部的に使用されるインターフェイスの実装となる、
  • subscribe の内部で Action1 をwrapして Subscriber のインスタンスを作成
  • その後 OnSubscribe の call を呼び出す。
    • これによって create の引数のラムダ式が呼び出される。
    • call の引数は直前に作成した Subscriber
  • ラムダ式は引数の Subscriber に対して、 onNext を呼び出す。
  • Subscriber の onNext は、 subscribe の引数のラムダ式をcallすることで呼び出す。

doOnNext

fun main(args: Array<String>) {
    Observable.create<String> { s ->
        s.onNext("one")
        s.onCompleted()
    }.doOnNext { i -> println("doOnNext $i") }
            .subscribe { s -> println("VALUE -> $s") }
}

オレンジが doOnNext 関連によって作成される処理
proxy的な動きをしていることがわかる

map

fun main(args: Array<String>) {
    Observable.create<String> { s ->
        s.onNext("one")
        s.onCompleted()
    }.map { "hello $it" }
            .subscribe { s -> println("VALUE -> $s") }
}

doOnNext とほぼ同じ動作

別ThreadからのPublisher

fun main(args: Array<String>) {
    println(Thread.currentThread())
    Observable.create<String> { s ->
        Thread {
            println(Thread.currentThread())
            s.onNext("one")
            s.onNext("two")
            s.onCompleted()
        }.start()
    }.subscribe { s -> println("VALUE -> $s by ${Thread.currentThread()}") }
}

// 出力結果
Thread[main,5,main]
Thread[Thread-0,5,main]
VALUE -> one by Thread[Thread-0,5,main]
VALUE -> two by Thread[Thread-0,5,main]
VALUE -> three by Thread[Thread-0,5,main]

別スレッドでpublishされた場合は、そのスレッドでsubscribeされる。