StateFlow と SharedFlow

StateFlow は Flow から状態の最新情報を適切に出力するための、また SharedFlow は Flow から値を複数のコンシューマに出力するための Flow API です。

StateFlow

StateFlow は、状態保持用の監視可能な Flow で、現在の状態や状態更新の情報をコレクタに出力します。現在の状態の値は、その value プロパティから読み取ることもできます。状態を更新してこの Flow に送信するには、MutableStateFlow クラスの value プロパティに新しい値を割り当てます。

Android において、StateFlow は状態を変更可能かつ監視可能に維持する必要のあるクラスに最適です。

Kotlin Flow の例に続けて、StateFlow を LatestNewsViewModel から公開し、View が UI 状態の更新をリッスンできるようにすることで、設定を変更しても画面の状態が正常に維持されるようにできます。

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    // Backing property to avoid state updates from other classes
    private val _uiState = MutableStateFlow(LatestNewsUiState.Success(emptyList()))
    // The UI collects from this StateFlow to get its state updates
    val uiState: StateFlow<LatestNewsUiState> = _uiState

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // Update View with the latest favorite news
                // Writes to the value property of MutableStateFlow,
                // adding a new element to the flow and updating all
                // of its collectors
                .collect { favoriteNews ->
                    _uiState.value = LatestNewsUiState.Success(favoriteNews)
                }
        }
    }
}

// Represents different states for the LatestNews screen
sealed class LatestNewsUiState {
    data class Success(val news: List<ArticleHeadline>): LatestNewsUiState()
    data class Error(val exception: Throwable): LatestNewsUiState()
}

MutableStateFlow の更新を担うクラスがプロデューサであり、StateFlow から収集を行うすべてのクラスがコンシューマです。flow ビルダーを使用して作成された「コールド」Flow とは異なり、StateFlow は「ホット」です。つまり、この Flow から収集してもプロデューサ コードはトリガーされません。StateFlow は常にアクティブでメモリに常駐しており、ガベージ コレクションの対象となるのはガベージ コレクション ルートからの参照が他にない場合のみです。

新しいコンシューマは、Flow からの収集を開始すると、ストリームの最後の状態とその後の状態を受け取ります。この動作は、LiveData などの他の監視可能なクラスと同様です。

View による StateFlow のリッスンは、他の Flow の場合と同様、次のように行います。

class LatestNewsActivity : AppCompatActivity() {
    private val latestNewsViewModel = // getViewModel()

    override fun onCreate(savedInstanceState: Bundle?) {
        ...
        // Start a coroutine in the lifecycle scope
        lifecycleScope.launch {
            // repeatOnLifecycle launches the block in a new coroutine every time the
            // lifecycle is in the STARTED state (or above) and cancels it when it's STOPPED.
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                // Trigger the flow and start listening for values.
                // Note that this happens when lifecycle is STARTED and stops
                // collecting when the lifecycle is STOPPED
                latestNewsViewModel.uiState.collect { uiState ->
                    // New value received
                    when (uiState) {
                        is LatestNewsUiState.Success -> showFavoriteNews(uiState.news)
                        is LatestNewsUiState.Error -> showError(uiState.exception)
                    }
                }
            }
        }
    }
}

Flow を StateFlow に変換するには、stateIn 中間演算子を使用します。

StateFlow、Flow、LiveData

StateFlow と LiveData は類似しています。どちらも監視可能なデータホルダー クラスであり、アプリ アーキテクチャでは同様のパターンで使用されます。

ただし、StateFlow と LiveData の動作は異なります。

  • StateFlow では初期状態をコンストラクタに渡す必要がありますが、LiveData ではその必要はありません。
  • LiveData.observe() では、ビューが STOPPED 状態になるとコンシューマが自動的に登録解除されますが、StateFlow またはその他の Flow からの収集では、収集は自動的には停止されません。同じ動作を実現するには、Lifecycle.repeatOnLifecycle ブロックから Flow を収集する必要があります。

shareIn を使用したコールド Flow のホット化

StateFlow は、「ホット」Flow です。つまり、収集されている間やガベージ コレクション ルートからの参照が他に存在する場合は、メモリ内に常駐します。コールド Flow をホットにするには、shareIn 演算子を使用します。

Kotlin Flow で作成した callbackFlow を例にとると、各コレクタで新しい Flow を作成する代わりに、shareIn を使用することで Firestore から取得したデータをコレクタ間で共有できます。それには、次の情報を渡す必要があります。

  • Flow の共有に使用する CoroutineScope。このスコープは、共有 Flow を必要な期間存続させるために、どのコンシューマよりも長く存続させる必要があります。
  • 新しいコレクタそれぞれに対してリプレイするアイテムの数。
  • 開始動作ポリシー。
class NewsRemoteDataSource(...,
    private val externalScope: CoroutineScope,
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        ...
    }.shareIn(
        externalScope,
        replay = 1,
        started = SharingStarted.WhileSubscribed()
    )
}

この例では、latestNews Flow は、最後に出力されたアイテムを新しいコレクタにリプレイし、externalScope とアクティブなコレクタが存在する限りアクティブな状態を維持します。SharingStarted.WhileSubscribed() 開始ポリシーは、アクティブなサブスクライバが存在する間は、アップストリーム プロデューサをアクティブに保ちます。他の開始ポリシーとして、プロデューサをすぐに開始する SharingStarted.Eagerly や、最初のサブスクライバが現れたときに共有を開始し、Flow を永続的にアクティブに保つ SharingStarted.Lazily などを利用できます。

SharedFlow

shareIn 関数からは SharedFlow が返されます。これはホット Flow で、そこからの収集を行うすべてのコンシューマに値を出力します。SharedFlow は、StateFlow を一般化して詳細な設定を可能にしたものです。

SharedFlow は、shareIn を使用せずに作成できます。たとえば、SharedFlow を使用すると、アプリの他の部分にティックを送信して、定期的にすべてのコンテンツをまとめて更新するといったことができます。最新ニュースの取得はもちろん、ユーザー情報セクションをお気に入りトピックのコレクションで更新することもできます。次のコード スニペットでは、TickHandler で SharedFlow を公開し、他のクラスがコンテンツ更新のタイミングを認識できるようにしています。StateFlow の場合と同様、クラス内で MutableSharedFlow タイプのバッキング プロパティを使用して、次のようにアイテムを Flow に送信します。

// Class that centralizes when the content of the app needs to be refreshed
class TickHandler(
    private val externalScope: CoroutineScope,
    private val tickIntervalMs: Long = 5000
) {
    // Backing property to avoid flow emissions from other classes
    private val _tickFlow = MutableSharedFlow<Unit>(replay = 0)
    val tickFlow: SharedFlow<Event<String>> = _tickFlow

    init {
        externalScope.launch {
            while(true) {
                _tickFlow.emit(Unit)
                delay(tickIntervalMs)
            }
        }
    }
}

class NewsRepository(
    ...,
    private val tickHandler: TickHandler,
    private val externalScope: CoroutineScope
) {
    init {
        externalScope.launch {
            // Listen for tick updates
            tickHandler.tickFlow.collect {
                refreshLatestNews()
            }
        }
    }

    suspend fun refreshLatestNews() { ... }
    ...
}

SharedFlow の動作は、次の方法でカスタマイズできます。

  • replay を使用すると、以前に出力された複数の値を新しいサブスクライバに再送信できます。
  • onBufferOverflow を使用すると、バッファが送信アイテムでいっぱいになったときのポリシーを指定できます。デフォルト値は BufferOverflow.SUSPEND で、呼び出し元を停止します。他のオプションには、DROP_LATEST と DROP_OLDEST があります。

MutableSharedFlow には、アクティブなコレクタの数を示す subscriptionCount プロパティもあり、それに応じてビジネス ロジックを最適化できます。また、MutableSharedFlow には resetReplayCache 関数も含まれており、Flow に送信された最新情報をリプレイしない場合に使用できます。

Flow に関する参考情報