RxKotlin

Tweet

Flowableのサンプル1

import io.reactivex.Flowable

fun main(args:Array<String>) {
    var flow = Flowable.just(1,2,3)
    flow.subscribe{
        println(it)
    }

    Thread.sleep(1*1000)
}

Flowableのサンプル2

import io.reactivex.*
import io.reactivex.schedulers.Schedulers
import org.reactivestreams.Subscriber
import org.reactivestreams.Subscription

//FlowableOnSubscribeは購読時にフロー可能(データ提供可能)という意味。
class MyFlow: FlowableOnSubscribe<String> {
    override fun subscribe(emitter: FlowableEmitter<String>) {
        //データ発射
        for(i in 0..100){
            emitter.onNext(i.toString())
        }

        //終わり
        emitter.onComplete()
    }
}

//データを購読する人
class MySubScriber<String>: Subscriber<String> {
    var subscription:Subscription? = null

    override fun onComplete() {
        println("終わりました")
    }

    override fun onSubscribe(s: Subscription?) {
        //一個だけデータください
        subscription = s
        subscription?.request(1)
        println("購読準備完了")
    }

    override fun onNext(t: String) {
        println("受信しました ${t}")
        subscription?.request(1)
    }

    override fun onError(t: Throwable?) {
        println("エラーが発生しました")
    }

    fun dispose() {
        subscription?.cancel()
    }
}


fun main(args:Array<String>) {

    //データ作成者のインスタンス生成
    var myFlow = MyFlow()

    //購読者のインスタンス生成
    var mySubscriber = MySubScriber<String>()

    //データ作成マシンを作成(myFlowはデータを作るパイロット、flowはパイロット搭載のデータ作成マシン)
    var flow = Flowable.create<String>(myFlow, BackpressureStrategy.BUFFER)

    //監視開始
    flow.observeOn(Schedulers.computation()) //別スレッドでmySubscriberを実行する
        .subscribe(mySubscriber)             //購読する人を登録する

    Thread.sleep(1*1000)
}