今回の話

【２．実装編】の続き

どんなところで使うんだろう

これまでRxで書くとこんな感じだよという説明を書いてきましたが、
実際のプロジェクトにおいてどのようなタイミングで使うかというと
大体２パターンくらいかなと思います。

非同期通信の時
ビーコンやセンサーなどのイベントを処理する時

今回はこの２パターンをRxで書く場合の手順を記載していきます。

非同期通信の時

Android標準で用意されている非同期処理(AsyncTask)がまぁ大変なので、非同期処理を行う際にRxを利用することが多いです。
【２．実装編】で記載している通り、まずはストリームを作成し、データソースを設定します。

ストリーム　　：単一のデータを一回だけ取ってくるのでSingleとする。
データソース　：通信APIをリクエストし、レスポンスが返ってくればストリームに流す。

repository.kt

    /**
     * 小説の内容を返却する
     */
    fun getBody() : Single<ResponseBody> {
        return Single.create {emitter->

            // novelBodyService:Retrofitで作成したAPIインターフェース
            val responseBody : Call<ResponseBody> = novelBodyService.getNovel("n8261fc", "1")

            responseBody.enqueue(object : Callback<ResponseBody> {
                override fun onResponse(call: Call<ResponseBody>, response: retrofit2.Response<ResponseBody>) {
                    // データの取得に成功したらストリームにデータを流す
                    // ※ObservableにおけるonNext()
                    emitter.onSuccess(response.body()!!)
                }

                override fun onFailure(call: Call<ResponseBody>, t: Throwable) {
                    emitter.onError(t)
                }
            })
        }
    }

上記ストリームができたら、監視を開始することでデータソースの処理が実行されます。
監視の開始方法はsubscribe APIを実行します。


   repository.getBody()
             .subscribe()

ただ、このままではResponseBodyインスタンス(html)が返却されるだけで何もしていません。
TextViewに表示できるように、ResponseBodyをテキストに変換します。


   repository.getBody()
             .flatMap { response -> translateResponseToText(response.toString()) }
             .subscribe()

こうするとsubscribeにはテキスト(String)が渡されますので
それをview.textに設定します。これで、APIから取得されたレスポンスの文字列をviewに設定する事ができました。
(translateResponseToTextはhtmlのタグを取り除いて文字列だけにする自作メソッドです)


   repository.getBody()
             .flatMap { response -> translateResponseToText(response.toString()) }
             .subscribe{text -> view.text = text}

しかしこのままだと、呼ぶ側のスレッドによって、どのスレッドで動作する関わるため、
メインスレッドで呼んでしまうと通信処理がメインスレッドで実施されてしまうため動きません。
そのためsubscribeされるまでの処理を別スレッドで動いてもらうよう、明示的に記載します。
※AndroidSchedulers.mainThread()を利用するにはRxAndroidが必要です。


   repository.getBody()
             .flatMap { response -> translateResponseToText(response.toString()) }
             .subscribeOn(Schedulers.io()) // subscribeされるまでは別スレッドで処理
             .observeOn(AndroidSchedulers.mainThread()) // 通知が来たらメインスレッドで処理
             .subscribe{ text -> view.text = text }

後は通信中にダイアログが出ていて欲しいので、
処理開始時にダイアログ表示、処理終了時にダイアログ非表示します。

MainActivity.kt


      repository.getBody()
                .flatMap { response -> translateResponseToText(response.toString()) }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe{showProgressDialog()} //処理開始時にダイアログ表示
                .doFinally{hideProgressDialog()} // 処理終了時にダイアログ非表示
                .doOnError { t -> Log.d("ERROR", t.message) }
                .subscribe { text ->
                    Log.d("GET NOVEL", "success")
                    view.text = text
                }

これで非同期で通信処理を実装できました！

イベントを処理する時

ビーコンのように大量にデータが飛んでくる時は、意味のあるデータだけ受け取って処理したいですよね。
そういったときもRxは便利です。

まず非同期処理と同じようにストリームを作成してデータソースを設定します。
ストリーム　　：データを何回も流したいのでObservableとする。
データソース　：Android のBluetoothをスキャンし、スキャン結果が設置ビーコンならストリームに流す。


    val stream = Observable.create<Beacon> { subscriber ->
        scanCallback = object : ScanCallback() {

            override fun onScanResult(callbackType: Int, result: ScanResult) {
                super.onScanResult(callbackType, result)
                Log.d("beacon", "onScanResult")
                if (result.scanRecord == null) return

                // scan結果をもとにbeaconクラス(自作)を
                val beacon = convertScanResultToBeacon(result.scanRecord.bytes, result.rssi)
                if (beacon != null && beacon.isSettingBeacon()) {
                    subscriber.onNext(beacon)
                }
            }

            override fun onBatchScanResults(results: List<ScanResult>?) {
                super.onBatchScanResults(results)
            }

            override fun onScanFailed(errorCode: Int) {
                super.onScanFailed(errorCode)
                Log.d("beacon", "scan failed")
                if (errorCode == 0) {
                    return
                }
            }
        }

        // APILevel21以上用
        mBluetoothLeScanner = mBluetoothAdapter!!.bluetoothLeScanner

        val scanSettings = ScanSettings.Builder().setScanMode(ScanSettings.SCAN_MODE_LOW_LATENCY).build()

        mBluetoothLeScanner!!.startScan(null, scanSettings, scanCallback)
    }!!

あとは非同期処理と同様にストリームを処理していきます。

beaconSubscription = stream
                .buffer(2000L, TimeUnit.MILLISECONDS) // 2秒間受信ビーコンを貯める
                .observeOn(AndroidSchedulers.mainThread()) // subscribeはメインスレッドで実行
                .subscribe({ beacons -> // 2秒間の間受信したビーコンを利用して処理実施})

Androidのイベント系の処理もRxで書けました！

まとめ

とりあえずRxで書いてみたい人向けに、書くために必要なことだけまとめてみました。
そのため使うことのメリット・デメリットについては記載してないです。

こんなところがわからなかったなど、
フィードバックいただければ幸いです。