kotlin-flows

Kotlin Flows

このリファレンスでは、Flow の落とし穴や意味上のエッジケース、つまり Channel と SharedFlow の違い、コールバックのブリッジ、再試行／エラー処理、変換処理内の副作用、KMP の境界について説明します。コールド／ホットストリームの基本、オペレーターの一覧、ライフサイクルに安全なコレクションについては扱いません。

Channel と SharedFlow — つまずきやすい意味上の違い

見つかったもの	対応
BroadcastChannel	SharedFlow に移行（非推奨）
ConflatedBroadcastChannel	StateFlow に移行（非推奨）
単一のコンシューマー向けの一度だけ発火するイベント（ナビゲーション、スナックバー、単発のエフェクト）に使う Channel	そのまま使う — 適切です。 Channel(BUFFERED).receiveAsFlow()
複数のコレクターへのブロードキャストに使う Channel	SharedFlow に移行（以下を参照）
プロデューサーとコンシューマーのキューとして使う Channel	そのまま使う — 適切です

Channel.receiveAsFlow() はファンアウトであり、ブロードキャストではありません。 複数のコレクターがいる場合、各イベントを受け取るのは1つのコレクターだけです（どのコレクターが受け取るかはフレームワークが選びます）。すべてのコレクターにすべてのイベントを届ける必要がある場合は、SharedFlow が必要です。これは、マルチキャストに Channel を使おうとしたときに陥りやすい落とし穴です。

単一のコンシューマー向けの一度限りのイベントには、Channel(BUFFERED).receiveAsFlow() をデフォルトとして使います。send は消費されるまで一時停止するため（イベントはキューに格納され、決して破棄されません）、一方、SharedFlow(replay = 0) は発行した瞬間にアクティブなコレクターがいないと、その発行を破棄します。公開するのは Flow だけにし、LaunchedEffect 内で collect を使って収集します。最後のイベントを状態として保持し、再コンポジションや設定変更のたびに再発火させる collectAsStateWithLifecycle は使わないでください。

private val _events = Channel<UiEvent>(Channel.BUFFERED)
val events: Flow<UiEvent> = _events.receiveAsFlow()

// in the UI
LaunchedEffect(Unit) {
    viewModel.events.collect { event -> /* navigate / snackbar — consumed once */ }
}

厳密には「一度だけ」ではなく、「非常に高い確率で最大一度」です。 receiveAsFlow() のコレクターが receive() の成功後、要素を処理するブロックの実行前にキャンセルされると、その要素は失われます。まれなケースです（ほとんどのキャンセルは受信と受信の間に発生します）が、決済や永続化に関わる重要なシグナルの場合は、結果を永続的な状態に保存してください（UiState の pendingResult フィールドに保存し、消費後に UI がクリアします）。compose/references/state-management.md の「Durable state + acknowledgement」を参照してください。

SharedFlow：tryEmit より launch { emit() }

デフォルトでは launch { _events.emit(e) } を使います。コレクターの準備ができるまで一時停止するため、エフェクトが気づかれないまま失われることはありません。デフォルトの MutableSharedFlow()（バッファなし）で tryEmit() を使うと、購読者が受信可能な状態でない場合に黙って破棄されます。extraBufferCapacity = 1 を追加しても、問題が起きるタイミングが先送りされるだけです（バッファがいっぱいのときに短時間で2回目の発行を行うと false が返され、エラーなしで破棄されます）。tryEmit は、取りこぼしても許容できるエフェクト（ツールチップ、サウンド）に限って使ってください。

読み取り専用 Flow の公開 — 明示的なバッキングフィールド（Kotlin 2.4+）

_x（可変）と x（読み取り専用）の2つのプロパティを使うイディオムは、明示的なバッキングフィールドを使った単一のプロパティにまとめられます。Kotlin 2.4 で安定版となり、2.3 では実験的機能です。

val uiState: StateFlow<UiState>
    field = MutableStateFlow(UiState())   // inside the class: uiState.update { … }; callers see StateFlow

これは、可変のバッキングフィールドから StateFlow／SharedFlow を公開する箇所に適用できます（MutableStateFlow は StateFlow でもあるため、デフォルトのゲッターで型チェックが通ります）。Channel を Flow として公開する場合には適しません。Channel は Flow ではないため、上記の明示的な _events／receiveAsFlow() のペアを使ってください。Kotlin < 2.4 の場合は（バージョンカタログで確認してください）、従来どおり private val _x と val x = _x.asStateFlow() を使ってください。

コールバックのブリッジ

// Single value: suspendCancellableCoroutine — always prefer it (propagates cancellation)
suspend fun authenticate(token: String): User = suspendCancellableCoroutine { cont ->
    val call = authApi.authenticate(token) { user, error ->
        if (user != null) cont.resume(user) else cont.resumeWithException(error ?: Exception())
    }
    cont.invokeOnCancellation { call.cancel() }
}
// suspendCoroutine only when the API has no cancellation concept at all.

// Stream: callbackFlow — awaitClose is MANDATORY (omitting it leaks the callback and never completes)
fun LocationManager.locationUpdates(provider: String): Flow<Location> = callbackFlow {
    val listener = LocationListener { trySend(it) }
    requestLocationUpdates(provider, 1000L, 0f, listener)
    awaitClose { removeUpdates(listener) }
}

登録解除 API のないサードパーティ SDK では、awaitClose 内でフラグを設定して終了を通知し、その制約をドキュメントに記載してください。

retry／retryWhen — 試行回数を必ず制限する

repository.getItems()
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) { delay((attempt + 1) * 1_000L); true } else false
    }
    .collect { ... }

attempt は0始まりの再試行インデックスです。attempt < による回数チェックがないと、永続的なエラーが発生したときにフローは無限に再試行します。（retry(3) { it is IOException } が簡潔な書き方です。）

エラー処理 — .catch の適用範囲とキャンセルの落とし穴

.catch が捕捉するのは上流のエラーだけです（collect {} 内で発生した例外は対象外です）。また、try/catch とは異なり、CancellationException は握りつぶしません。そのため、オペレーター形式を優先し、コレクター側の処理では具体的な型の例外を捕捉してください。

// ✓ upstream errors, cancellation-safe
repository.getItems().onEach { process(it) }.catch { logger.error(it) }.collect()

// ✗ catch (e: Exception) inside collect/any suspend fn swallows CancellationException and breaks
//   structured concurrency — the single most common coroutine bug.
//   If a broad catch is unavoidable: if (e is CancellationException) throw e  // before handling

combine／map の変換処理の外で副作用を実行する

変換ラムダは純粋であり、上流から値が出力されるたび、また再購読されるたびに再実行されます（例：WhileSubscribed によるローテーション）。その中に viewModelScope.launch {} やイベントの出力を記述すると、繰り返し実行されます。変換処理の後にある onEach に移してください。

combine(userFlow, settingsFlow) { user, settings -> UiState(user, settings) }
    .onEach { analytics.logView(it.user.id) }   // not inside combine
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), UiState.Empty)

combine は、最初の出力を行う前にすべての入力から少なくとも一度値が出力されるのを待ちます。 入力のひとつが一度も値を出力しないコールドフローの場合、結合されたフローも値を出力しません。これは「画面が Loading のまま止まる」という典型的なバグです。各入力を StateFlow にするか、コールド入力に onStart { emit(initial) } を指定するか、センチネル値を先頭に追加してください。（combine(a, b) { (x, y) -> … } の分割代入はコンパイルできません。）

KMP

共有コードから Flow を公開し、プラットフォームごとに収集してください（iOS：SKIE または CoroutineScope ラッパー）。コルーチン外の iOS コンテキストから StateFlow.value を読み取るのは避け、収集用ラッパーを使用してください。

テスト

Flow のテストについては android-skills:android-testing を参照してください。ホットフローを検証する前に、2 つのスケジューラーに関する落とし穴（runTest と MainDispatcherRule は同じスケジューラーを共有する必要があります）を確認してください。