ロコガイドで Android エンジニアをしております、横山です。
みなさん、Android プロジェクトでのKotlin Coroutine Flowの活用は進んでおりますでしょうか。
Android の Developers サイトでも丁寧なガイドが公開されていますが、
いざ自社プロダクトに取り入れようとしたときに、細かなユースケースでどうすりゃええんや…と悩める場面がいくつかあったので、
その中から一例を取り上げて紹介させていただきます。
この記事の対象読者
- Android アプリケーション開発者の方
- Kotlin Coroutine Flow に触れたことがある
- 全くの初学者の方は先に、Android Developers ガイドを見るのをおすすめします。
やりたいこと
今回紹介するのは以下のような手動操作でのリトライ処理です。
ありふれたユースケースですが、Flow は複数の値を出力するデータストリームである故に、
任意のタイミングで処理を再開するのに少し苦労したので、その解決方法を簡単なサンプルコードとともに紹介します。
DataSource/Repository
まず大元のデータストリームですが、
Android での Kotlin Flowのサンプルコードをベースに、一定間隔で値を emit する Flow を用意しました。
class MyDataSource { val latest: Flow<ApiResponse> = flow { while (true) { delay(3000) // ランダムにExceptionをthrow if (Random.nextBoolean()) { throw Exception("") } emit(ApiResponse(message = "success")) } } } class MyRepository( dataSource: MyDataSource, ) { val latest: Flow<ApiResponse> = dataSource.latest }
エラーを再現しやすいように、ランダムに例外をスローしています。
ViewModel
ViewModel では RepositoryFlow の値を View に適した形へと変換し、
最終的にはasLiveData関数を用いて LiveData 化することで、View でより取り扱いやすいようにしています。
class ManualRetryViewModel : ViewModel() { ... val uiState: LiveData<UiState> = repository.latest .map { UiState.Success(it.message) as UiState } .catch { emit(UiState.Error) } // エラーハンドリング .asLiveData() // FlowをLiveDataに変換 } sealed class UiState { class Success(val message: String) : UiState() object Error : UiState() }
catch
RepositoryFlow からスローされる例外をキャッチするために、途中catch
演算子を用いています。
ref: 予期しない例外のキャッチ
しかし、このコードには問題があります。
catch
演算子で例外処理を行うと、その時点でストリームが完全停止し、以降のリトライができなくなるからです。
catch
演算子の内部実装をのぞいてみましょう。
public fun <T> Flow<T>.catch(action: suspend FlowCollector<T>.(cause: Throwable) -> Unit): Flow<T> = flow { val exception = catchImpl(this) if (exception != null) action(exception) }
関数内で新しい Flow を生成していることがわかります。
さらに Flow 内で実行されているcatchImpl
の実装をみると・・
internal suspend fun <T> Flow<T>.catchImpl( collector: FlowCollector<T> ): Throwable? { var fromDownstream: Throwable? = null try { collect { try { collector.emit(it) } catch (e: Throwable) { fromDownstream = e throw e } } } catch (e: Throwable) { ... // 省略 }
(catch
演算子の)呼び出し元である Flow に collect し、出力された値を新しく生成した Flow に emit していることがわかります。
また、スローされた例外はここで try-catch されています。
例外をキャッチしたあとは、呼び出し元の Flow に再 collect されないので、ストリームが完全に停止することになります。
retryWhen
catch 演算子の他に例外を処理する方法として、retryWhen
演算子があります。
class ManualRetryViewModel : ViewModel() { ... val uiState: LiveData<UiState> = repository.latest .map { UiState.Success(it.message) as UiState } .retryWhen { cause, attempt -> emit(UiState.Error) true } // エラーハンドリング .asLiveData() }
cause
はスローされた例外、attempt
は現行のリトライ回数です。
例外の内容やリトライ回数に応じて再リトライするかを判定し、その結果を Boolean で返します。
こちらも内部実装をみると・・
public fun <T> Flow<T>.retryWhen(predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean): Flow<T> = flow { var attempt = 0L var shallRetry: Boolean do { shallRetry = false val cause = catchImpl(this) if (cause != null) { if (predicate(cause, attempt)) { shallRetry = true attempt++ } else { throw cause } } } while (shallRetry) }
catch 演算子と同じく、内部で新しい Flow を生成し、catchImpl
を実行しています。
catch との違いは、do..while でループ処理をしていることです。
predicate
の結果が true のとき、例外がスローされた後に再びcatchImpl
が呼ばれ、呼び出し元の Flow に再接続されます。
これによりストリームの自動リトライが可能です。
これで完璧なようにも思えますが、今回の目的は手動リトライなのでもう一工夫必要になります。
manual retry
まずリトライ処理自体をイベントクラスとして定義し、
さらにイベント状態を送出するデータストリームを作成しました。
class ManualRetryViewModel : ViewModel() { ... private val retryEvent = MutableLiveData<RetryEvent>(RetryEvent.Retrying) val uiState: LiveData<UiState> = retryEvent.asFlow() ... fun retry() { retryEvent.value = RetryEvent.Retrying } } sealed class RetryEvent { object Retrying : RetryEvent() object Retried : RetryEvent() }
イベントの状態は View でも監視したかったので、
イベントソースは LiveData で生成し、asFlow関数で Flow に変換しています。
Flow 単体で同様の処理を行う場合は、MutableStateFlow
またはMutableSharedFlow
を使うことで、
LiveData と同様に、任意のタイミングで値を送出可能です。
ref: StateFlow と SharedFlow
private val retryState = MutableStateFlow<RetryEvent>(RetryEvent.Retrying) fun retryWithStateFlow() { retryState.value = RetryEvent.Retrying }
リトライイベントの Flow を作成したら、
flatMapLatest
演算子を用いて、RepositoryFlow を連結します。
こうすることで、リトライイベント Flow の値出力を待ってから RepositoryFlow が開始するようになります。
class ManualRetryViewModel : ViewModel() { ... val uiState: LiveData<UiState> = retryEvent.asFlow() .filter { it is RetryEvent.Retrying } // リトライ中だけ値を送出 .flatMapLatest { repository.latest } .map { UiState.Success(it.message) as UiState } .retryWhen { _, _ -> emit(UiState.Error) true } .onEach { retryEvent.value = RetryEvent.Retried } .asLiveData() }
この状態で RepositoryFlow から例外がスローされると、
retryWhen内のFlowで再connect ↓ リトライイベントFlowが再び値を待機 ↓ (retry関数を呼び出して)リトライイベントFlowに値をemit ↓ RepositoryFlowが再開
となり、手動でのリトライが実現可能となります。
View
ストリームのコンシューマーである Activity のコードは以下の通りです。
class ManualRetryActivity : AppCompatActivity() { private lateinit var binding: ActivityManualRetryBinding private val viewModel: ManualRetryViewModel by viewModels() override fun onCreate(savedInstanceState: Bundle?) { ... binding.retry.setOnClickListener { viewModel.retry() } viewModel.retryEvent.observe(this) { when (it) { is RetryEvent.Retrying -> showProgress() is RetryEvent.Retried -> hideProgress() } } viewModel.uiState.observe(this) { when (it) { is UiState.Success -> showMessage(it.message) is UiState.Error -> showError() } } } }
終わりに
いかがでしたでしょうか。
サンプルコードの全容は ↓ で公開しているので合わせてご確認ください。
https://github.com/yokomii/AndroidCoroutineFlowManualRetry
ロコガイドでは業務で Coroutine Flow をバリバリ書きたい Android エンジニアを随時募集しています。
ご興味のあるかたはお気軽にお問い合わせください ↓↓
Android アプリケーション開発エンジニア/MAU800 万人・国内最大級の買い物情報サービス