ちょっと便利にするRx

割とよく使う小技。

リジューム可能なsubcribe

一度、onErrorが呼ばれるとそのストリームに値を流せません。(すでにエラーが起きているよ。と値を流すと例外が飛んでくる)
イベントハンドラ的にobservableを使っていると、一度エラーが起きても復帰させて継続的にイベントを受けたいような時があると思います。
エラー → dispose → 再度subscribeなんていちいちやってられないので、そんな時は、一度 materialize すれば大丈夫です。

// T型を受けるコールバック関数
typealias OnNext<T> = (T) -> Unit
typealias OnError = (Throwable) -> Unit
typealias OnComplete = () -> Unit

@CheckReturnValue
fun <T> Observable<T>.resumeSubscribeBy(
    onError: OnError? = null,
    onComplete: OnComplete? = null,
    onNext: OnNext<T>? = null
): Disposable {
    return materialize()
        .subscribe {
            when {
                it.isOnError -> onError?.invoke(it.error!!)
                it.isOnNext -> {
                   // 例外時はonErrorに渡す
                    try {
                        onNext?.invoke(it.value!!)
                    } catch (e: Throwable) {
                        onError?.invoke(e)
                    }
                }
                it.isOnComplete -> onComplete?.invoke()
            }
        }
}

Notificationと合わせて使う

処理内容に応じて、通知を出し分けたい！
みたいなときありますよね。

拡張メソッドを利用して、以下な感じで。
notificationにしていますが、timberとかに変えてログを残すだけでもおすすめ。
特にスタックをログに残すようにすれば、どの処理かの特定がしやすくなります。

fun <T> Single<T>.withNotification(context: context, notificationBuilder: NotificationCompat.Builder, notificationContent: String): Single<T> {
// val stack = Throwable() // スタックトレース用

    return this.doOnSubscribe {
            notificationBuilder.setContentText("${notificationContent}の処理開始")

            val notificationManager = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager
            notificationManager.notify(notificationId, notificationBuilder.build())
        }
        .doOnError { err ->
            notificationBuilder.setContentText("${notificationContent}の処理中にエラーが発生しました")

            val notificationManager = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager
            notificationManager.notify(notificationId, notificationBuilder.build())
        }
        .doOnSuccess {
            notificationBuilder.setContentText("${notificationContent}のを実行しました")

            val notificationManager = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager
            notificationManager.notify(notificationId, notificationBuilder.build())
        }
}

リトライ

リトライ回数とリトライ間隔を調整したい！
みたな時はよくあるケースです。

flowable
    .take(10) // リトライ数
    .flatMapSingle { err ->
        // 間隔
        return@flatMapSingle Single.timer(1000, TimeUnit.MILLISECONDS)
    }

分岐

分岐して実行されるストリームを変える。みたいな時。flatMapを使いましょう！

val a = Random.nextInt(10)

Observable.just(a)
    .flatMap {
        if(it == 1) {
            return@flatMap Observable.just("a は 1です！！")
        }
        else {
            return@flatMap Observable.just("1じゃなかったーーー")
        }
    }
    .subscribe {
        Timber.d("結果:${it}")
    }