2017-07-12

How to convert Completable to Observable? Rxjava2

Observableな川の流れの中で、Completableを使った時に、どうやってまたObservableとして流せばいいのか？と考えてた。

http://reactivex.io/RxJava/javadoc/io/reactivex/Completable.html#andThen(io.reactivex.ObservableSource)

andThenというオペレータを使えばよさそう。

具体的なコードはこんな感じ

        Observable.just(1, 2, 3)
                .flatMap { number ->
                    Completable.fromAction {
                        showMessage("completableTest: " + number)
                        Thread.sleep(300)
                    }.andThen(Observable.just(number))
                }
                .doOnNext {
                    showMessage("doOnNext")
                }
                .doOnComplete {
                    showMessage("doOnComplete")
                }
                .subscribe({
                    showMessage("onNext: " + it.toString())
                }, {
                    it.printStackTrace()
                    showMessage("hoge")
                }, {
                    showMessage("onComplete")
                })

実行結果はこんな感じ

main:completableTest: 1
main:doOnNext
main:onNext: 1
main:completableTest: 2
main:doOnNext
main:onNext: 2
main:completableTest: 3
main:doOnNext
main:onNext: 3
main:doOnComplete
main:onComplete

※mainはスレッドの名前を出しているだけ

どう書いたらいいかわからなかったので、いろいろと試行錯誤してた

ダメなパターン１

.toObservable<Int>()でObservableに変換するだけのコード

        Observable.just(1, 2, 3)
                .flatMap { number ->
                    Completable.fromAction {
                        showMessage("completableTest: " + number)
                        Thread.sleep(300)
                    }.toObservable<Int>() // ここが変わっただけ
                }
                .doOnNext {
                    showMessage("doOnNext")
                }
                .doOnComplete {
                    showMessage("doOnComplete")
                }
                .subscribe({
                    showMessage("onNext: " + it.toString())
                }, {
                    it.printStackTrace()
                    showMessage("hoge")
                }, {
                    showMessage("onComplete")
                })

実行結果

main:completableTest: 1
main:completableTest: 2
main:completableTest: 3
main:doOnComplete
main:onComplete

Completableの中身はちゃんと実行されているけど、onNextには何も値が流れてこない。

ダメなパターン２

toObservableからの、startWith

        Observable.just(1, 2, 3)
                .flatMap { number ->
                    Completable.fromAction {
                        showMessage("completableTest: " + number)
                        Thread.sleep(300)
                    }.toObservable<Int>()
                            .startWith(number)
                }
                .doOnNext {
                    showMessage("doOnNext")
                }
                .doOnComplete {
                    showMessage("doOnComplete")
                }
                .subscribe({
                    showMessage("onNext: " + it.toString())
                }, {
                    it.printStackTrace()
                    showMessage("hoge")
                }, {
                    showMessage("onComplete")
                })

実行結果

main:doOnNext
main:onNext: 1
main:completableTest: 1
main:doOnNext
main:onNext: 2
main:completableTest: 2
main:doOnNext
main:onNext: 3
main:completableTest: 3
main:doOnComplete
main:onComplete

一見、onNextも呼ばれてそうだし、大丈夫そう！と思うのだが、順番がおかしい。川の流れ的に、completableTestが呼ばれてからonNextが呼ばれてほしいが、逆になってしまっている。

よさげなパターン

一度singleに変換してobservableに変換したら上手く言った。上手く言ったのでCompletableのところだけ貼る。

                    Completable.fromAction {
                        showMessage("completableTest: " + number)
                        Thread.sleep(300)
                    }
                            .toSingleDefault(number)
                            .toObservable()

Conclusion

川を使いこなすにはまだまだ修行が必要だー

雑なサンプル置き場を作ってる https://github.com/mapyo/RxJavaSamples/blob/master/app/src/test/java/com/mapyo/rxjavasamples/RxExampleUnitTest.kt

2017-07-05

Pythonでバイナリを保存する時

最初は

        f = open('hogedata','w')
        for packet in packets:
            f.write(packet)
        f.close()

こう書いてた。Macだと上手くいくのだけど、Windowsで試して貰った時にダメだった。ダメだったというのは、Macで作った時と何故か違うファイルが出来ているっぽかった。ファイルサイズもそもそも違うし、出来たファイルを後工程で使うものに渡してもエラーで動かない。

あれなんでだろ。。？と思っていろいろ調べていくと、

https://docs.python.jp/2.7/library/functions.html?highlight=open#open

バイナリとテキストファイルを区別するシステムでは、ファイルをバイナリモードで開くためには 'b' を追加してください; 区別しないシステムでは 'b' は無視されます。

あと、毎回新しくファイルを作り直したかったので、w+にした

        f = open('hogedata','w+b')
        for packet in packets:
            f.write(packet)
        f.close()

バイナリをなんかする時はちゃんとバイナリモードでファイルを開きましょう。

それにしても、Mac上でもうまくいかないでほしい。。けど、Wndowsとなにかしら違うからそういうもんなのかな？（雑）

2017-07-02

別プロセスで動いている生き死にするServiceで非同期処理をする事を考える

AndroidのServiceのお話。以下のようなServiceがある。

Serviceを別プロセスで動かしている
destroyが呼ばれた時は、Process.killProcess(Process.myPid())を呼んで自分自身のプロセスをkillする。
定期的に生き死にする

こういったSerivceで非同期処理をすると、ちゃんと非同期処理の終了を待つ処理をいれないと、非同期処理してる間にサービスが死しに、プロセス自体が死ぬので非同期処理が完了する前に終わってしまう。。。

みたいな事があります。

なので、別プロセスのSerivceのライフサイクルに依存しない形で非同期処理をしたい。

IntentServiceを使ってアプリのプロセスで動かすようにすると、管理しやすいのでないか？と思ったので実際にコードを書いて試してみます。また、Serviceの中で非同期処理して、見事に途中で終わってるよね。という事も確認したかったので、それも試してみています。

実験した内容は以下の２つ

何も考えずにServiceの中で非同期処理をする
アプリのプロセスでIntentServiceを呼んで非同期処理する

実験するServiceの仕様

Serviceが起動する
5秒たったらstopServiceを呼んで停止する
onDestroyが呼ばれた時にProcess.killProcess(Process.myPid())を呼んで自分自身をkillする

以下のようなコードをSerivceのonCreateに入れて、５秒たったら停止するようにする

        Handler(Looper.getMainLooper()).postDelayed({
            showMessage("stop service")
            this.stopService(Intent(this, SampleService::class.java))
        }, 5_000)

1. 何も考えずにServiceの中で非同期処理する

以下のような感じで1秒毎にメッセージを表示して、10秒たったら終了。みたいな処理をServiceのonCreateの中に書く。

        var counter = 0
        timer.schedule(object : TimerTask() {
            override fun run() {
                counter++
                showMessage("counter: " + counter)
                if (counter == 10) {
                    Handler(Looper.getMainLooper()).post {
                        showMessage("counter finished")
                        timer.cancel()
                    }
                }
            }

        }, 1_000, 1_000)

結果

counter: 1
counter: 2
counter: 3
counter: 4
stop service
destroy service

Serviceは5秒で停止する。onDestoryが呼ばれた時に自分自身をkillしてるので、counterは１０秒までは回らない。まぁ、プロセスが死ぬんだから、その通りだ。

2. アプリのプロセスでIntentServiceを呼んで非同期処理する

こんな感じでAndroidManifest上で以下のように書きます。Serviceを別プロセスで動かして、IntentServiceをアプリのプロセスで動かす。という書き方。

        <service android:name=".SampleService"
            android:process=":SampleService" />

        <service android:name=".SampleIntentService" />

https://developer.android.com/guide/topics/manifest/service-element.html

android:processの仕様はこの辺に書いてある。

processを書かない場合はアプリのプロセスで動くと書いてあって、ちゃんとアプリのプロセスで動くよね？呼び出し元のプロセスで動かないよね？という事も確認したかった。

実験したIntentServiceの仕様

SampleIntentServiceというIntentServiceのクラスを作って、onHandleIntentの中で以下のように書きます。

        (1..15).map {
            Thread.sleep(1000)
            showMessage("SampleIntentService counter: " + it)
        }

こんな感じで、１秒毎にカウンターを表示している。

結果

IntentServiceの方は最後までカウンターが表示されていた。
プロセスもちゃんとアプリのプロセスで動いてた。
別プロセスで動いているServiceの方は普通に起動して５秒後に死んでいた。

最後に

実験に使ったコードはこちらにアップしています。 https://github.com/mapyo/ProcessSample

IntentServiceを使うと、処理が終わったら勝手に終了してくれるし、連続で呼んでしまったとしても、逐次実行してくれるようになってるし、いろいろと便利です。

2017-06-28

RxJavaでretryWhenを使ってエラーをキャッチして別のエラーを流す

https://speakerdeck.com/yuyakaido/droidkaigi-2017?slide=86

この辺の話に近い。

Observable.create<Int> { emitter ->
    emitter.onNext(1)
    emitter.onNext(2)
    emitter.onError(HogeException())
    emitter.onComplete()
}.retryWhen { observable ->
    observable.flatMap { e ->
        val exception = if (e is HogeException) FooException() else e
        Observable.error<Int>(exception)
    }
}.subscribe({
    println(it)
}, { e ->
    println(e.javaClass.simpleName)
}, {})

class HogeException : RuntimeException()
class FooException : RuntimeException()

こんな感じでHogeExceptionが流れてきたら、FooExceptionにExceptionを変更して流す。

そんな１日だった。いや違うけど。

2017-06-12

RxJavaのdoAfterNextを使う

RxJavaで一つ前の川で受信用のobservableをflatmapとかで設定して、その直後で送信処理するみたいな事書きたいけど出来るのかな。。？逆にすると送信処理した直後にデータ送られて来たら、それを取りこぼしちゃうみたいな。。。
— ぼいらー (@mapyo) 2017年5月24日

こういう事がしたかった。

http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#doAfterNext(io.reactivex.functions.Consumer)

doAfterNextを使うと解決出来そうだった。サンプルコードはこちら publishSubject.onNextのとこらへんで送信開始処理みたいなのを行うイメージ

        val publishSubject = PublishSubject.create<Int>()

        Observable.just(1)
                .doAfterNext {
                    println("doAfterNext")
                    publishSubject.onNext(it * 2)
                }
                .flatMap {
                    println("receive start")
                    publishSubject
                }
                .subscribe({ println(it) })

出力結果はこちら

receive start
doAfterNext
2

flatMapでpublishSubjectが設定される
その後doAfterNextが呼ばれて、publishSubjectで1 * 2が流される
subscribeしてるところで２がとれる

という事で無事に出来ました。もうちょっと早く気がついてたらもっといい川を書けてたかもしれない。

2017-05-17

株式会社スマートドライブに入社して4ヶ月くらいがたった

気がついたら３ヶ月が過ぎて試用期間が終わって、４ヶ月過ぎようとしているのでもろもろ書いてみる

何やってる会社なの？

ざっくり言うと車に専用のハードウェアを取り付けて走行データを収集して、スマホ経由でサーバに送ってサーバで解析して、見やすい形にしてブラウザで表示するというサービスをやってる。

車から走行データを収集するためのハードウェア開発、ハードウェアで吸い上げた情報をBluetoothでスマホに送り、スマホからサーバに情報を送る部分や、情報を表示するためアプリ開発。サーバ側で各種情報を受け取って解析する部分や各種APIを開発しているサーバサイド開発、解析した結果を見やすい形で表示するフロントエンド周り。

ハードからアプリ、サーバ、フロントエンドといろんな事をやってる会社です。

ハード周りはあまりちゃんとわかってないけど、ファームウェアはC言語で書かれてるっぽい。サーバサイドはRailsとGo。フロントエンドはSPAでReactとか使ってるっぽい。

かなりざっくり書いたけど、もうちょっと詳しく知りたい方は以下のページをさらっと見てもらうのがいいと思います。

blog.smartdrive.co.jp

www.smartdrive.co.jp

www.wantedly.com

jp.techcrunch.com