SharedFlow is not collecting from emission

Question

In my ViewModel, I am making API requests and I am using StateFlow and SharedFlow to communicate with the Fragment. While making the API request, I am easily able to update the state flow's value and it is successfully collected in the Fragment.

But before making the request, I am emitting some boolean values with SharedFlow and it is not getting collected in the Fragment. Can someone help me why is this happening?

class MainViewModel: ViewModel() {
  private val _stateFlow = MutableStateFlow(emptyList<Model>())
  val stateFlow = _stateFlow.asStateFlow()

  private val _loading = MutableSharedFlow<Boolean>()
  val loading = _loading.asSharedFlow()

  suspend fun request() {
    _loading.emit(true)
    withContext(Dispatchers.IO) {
      /* makes API request */
      /* updates _stateFlow.value */
      /* stateFlow value is successfully collected */
    }
    _loading.emit(false) // emitting boolean value
  }
}

class MyFragment : Fragment(R.layout.fragment_my) {
   // ...

  override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    lifecycleScope.launchWhenStarted {
      viewModel.request()
      /* stateFlow is collected and triggered - working properly */

      viewModel.loading.collectLatest { // <- NOT COLLECTING - WHY?
        Log.d(this::class.simpleName, "onViewCreated: $it") // <- NOT LOGGING
      }
    }
  }
}

Why did you use `lifecycleScope.launchWhenStarted` instead of `onStart()`? — Barry Fruitman, Dec 08 '22 at 01:55

score 7 · Answer 1 · answered Jun 27 '22 at 21:50

SharedFlow is a hot stream. Probably you should create it with

MutableSharedFlow(
    replay = 0,
    onBufferOverflow = BufferOverflow.DROP_OLDEST,
    extraBufferCapacity = 1
)

or

MutableSharedFlow(
    replay = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

Sergio · Answer 2 · 2022-04-04T10:49:27.603

I guess you need to launch a different coroutine to collect loading values, something like the following:

 override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    lifecycleScope.launchWhenStarted {
      viewModel.request()
    }
    lifecycleScope.launchWhenStarted {    
      viewModel.loading.collectLatest { 
        Log.d(this::class.simpleName, "onViewCreated: $it")
      }
    }
  }

viewModel.request() function is a suspend function, it suspends the coroutine until it is finished. But I guess it is not finishing due to calling suspend function _loading.emit(), suspending until it is collected.

Or I think it is even better would be to launch a coroutine in ViewModel class, something like the following:

// In MainViewModel
fun request() = viewModelScope.launch {
    _loading.emit(true)
    withContext(Dispatchers.IO) {
      /* makes API request */
      /* updates _stateFlow.value */
      /* stateFlow value is successfully collected */
    }
    _loading.emit(false) // emitting boolean value
}

// In MyFragment
override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    lifecycleScope.launchWhenStarted {
      viewModel.loading.collectLatest { 
        Log.d(this::class.simpleName, "onViewCreated: $it") 
      }
    }

    viewModel.request()
}

And the reason you need separate coroutines is that collecting a StateFlow (always infinite) or any other infinite Flow in a coroutine means the `collect` call will never return because the flow is never done emitting values. — Tenfour04, Apr 04 '22 at 12:06
@Tenfour04 I would like to understand why the flow would not emit the value. — Sagar Patel, Aug 08 '23 at 22:54
@SagarPatel Do you mean in the code in the question at the top? They created a SharedFlow that has no replay. `request()` returns before they even start collecting from the SharedFlow. Since the SharedFlow has no replay set, it will not play values back for the new collector that were emitted before collection began. — Tenfour04, Aug 09 '23 at 01:39

Weidian Huang · Answer 3 · 2022-10-25T19:18:49.737

SharedFlow by default doesn't replay any value when you subscribe to it.

   viewModel.request() // _loading.emit() has executed before collection.

   viewModel.loading.collectLatest { // No value can be replayed here
        Log.d(this::class.simpleName, "onViewCreated: $it") 
   }

So for the above code, if want to collect the last 2 values when you subscribe to it. You can do something like this.

private val _loading = MutableSharedFlow<Boolean>(replay = 2)

But you won't to see the loading effect, because 2 values will be collected at the same time when you subscribe to it. Probably this is what you want, _loading is emitted in IO thread, so the request() can be immediately returned before _loading is set, and you can start to collect the value at Fragment level:

suspend fun request() {
    withContext(Dispatchers.IO) {
      _loading.emit(true)
      /* makes API request */
      /* updates _stateFlow.value */
      /* stateFlow value is successfully collected */
      _loading.emit(false) // emitting boolean value
    }
  }

StateFlow is similar to LiveData, it can emit the last value when the new subscriber subscribes to it. Something like SharedFlow with replay = 1. That's why you can still collect the last value in the above code.

SharedFlow is not collecting from emission

3 Answers3