はじめに

interface OnChangeListener {
    fun onChange()
}

のようなコールバックを、Flowに変換するにはどうすればいいかを調べました。

RxJavaでいうところ、Observable.create のようなことを実現するにはどうすればいいかという記事です。

結論

callbackFlowというAPIがcoroutineに用意されてるので、それを使うと良さそうです。

これは、cold flowを生成します。つまり、collectされるまで動きません。

サンプルを書いてみました。

flowFlow関数で、callbackFlowを使ってコールバックからFlowに変換しています。 OnChangeListener#onChangeの実装でofferを使用してチャンネルに追加しています。容量がいっぱいの場合は例外がなげられるとのこと。

awaitClose でunregsiterやcancel処理を実装する必要があります。これを実装しなければ、実行時に

IllegalStateException: ‘awaitClose { yourCallbackOrListener.cancel() }’ should be used in the end of callbackFlow block.

というエラーがでます。（感想ですが、クローズ処理を忘れないので、良い仕組みだと思います！）

callbackからflowに変換できたので、あとはcollectしてあげるだけです。このサンプルでは、1秒ごとにログに値が出力されます。

onchange: 1
onchange: 2
onchange: 3
onchange: 4
onchange: 5
onchange: 6
onchange: 7
...