[FIXED] Flow not emitting items

Issue

The code below implements a simple screen that shows a list of users using the MVI pattern with kotlin coroutines. The code runs but does not produce any results. After alot debugging, i found out that the getUsers function in the UserListVM does not emit anything. Would appreciate some help here. thanks alot.

    class UserListVM : ViewModel() {

    val resultFlows: Channel<Flow<*>> = Channel(Channel.UNLIMITED)
    val liveState = MutableLiveData<PModel<*, UserListIntents>>()
    val intents: Channel<UserListIntents> = Channel()

    lateinit var job: Job
    lateinit var currentState: UserListState

    fun offer(event: UserListIntents) = intents.offer(event)

    suspend fun store(initialState: UserListState): LiveData<PModel<*, UserListIntents>> {
        job = viewModelScope.launch {
            currentState = initialState
            intents.consumeEach { intent ->
                resultFlows.send(reduceIntentsToResults(intent, currentState)
                        .flowOn(Executors.newFixedThreadPool(4).asCoroutineDispatcher())
                        .map { SuccessResult(it, intent) }
                        .catch { ErrorEffectResult(it, intent) }
                        .onStart { emit(LoadingEffectResult(intent)) }
                        .distinctUntilChanged()
                )
            }
            resultFlows.consumeEach { results ->
                results.flatMapMerge {
                    val states = stateStream(this as Flow<Result<UserListResult, UserListIntents>>, currentState)
                    val effects = effectStream(this as Flow<Result<UserListEffect, UserListIntents>>)
                    flowOf(states, effects)
                }
                        .flattenMerge()
                        .collect { pModel -> liveState.value = pModel }
            }
        }
        job.start()
        return liveState
    }

    private suspend fun reduceIntentsToResults(intent: UserListIntents, currentState: Any): Flow<*> {
        Log.d("UserListVM", "currentStateBundle: $currentState")
        return when (intent) {
            is GetPaginatedUsersIntent -> when (currentState) {
                is EmptyState, is GetState -> getUsers()
                else -> throwIllegalStateException(intent)
            }
            is UserClickedIntent -> when (currentState) {
                is GetState -> flowOf((SuccessEffectResult(NavigateTo(intent.user), intent)))
                else -> throwIllegalStateException(intent)
            }
        }
    }

    private suspend fun getUsers(): Flow<UsersResult> {
        return flow {
            emit(UsersResult(listOf(User("user1", 1), User("user2", 2), User("user3", 3))))
        }.flowOn(Dispatchers.IO)
    }

    override fun onCleared() {
        super.onCleared()
        job.cancel()
    }
    }

    class UserListActivity : AppCompatActivity() {

    var intentStream: Flow<UserListIntents> = flowOf()
    lateinit var viewModel: UserListVM
    lateinit var viewState: UserListState
    private lateinit var usersAdapter: GenericRecyclerViewAdapter

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        initialize()
        setupUI(savedInstanceState == null)
        viewModel.store(viewState).observe(this, Observer {
            it?.apply {
                when (this) {
                    is ErrorEffect -> bindError(errorMessage, error, intent)
                    is SuccessEffect -> bindEffect(bundle as UserListEffect)
                    is SuccessState -> {
                        (bundle as UserListState).also { state ->
                            viewState = state
                            bindState(state)
                        }
                    }
                }
                toggleLoadingViews(intent)
            }
        })
    }

    fun initialize() {
        viewModel = getViewModel()
        viewState = EmptyState()
    }

    fun setupUI(isNew: Boolean) {
        setContentView(R.layout.activity_user_list)
        setSupportActionBar(toolbar)
        toolbar.title = title
        setupRecyclerView()
    }

    override fun onResume() {
        super.onResume()
        if (viewState is EmptyState) {
            GlobalScope.launch {
                viewModel.offer(GetPaginatedUsersIntent(0))
            }
        }
    }

    private fun bindState(successState: UserListState) {
        usersAdapter.setDataList(successState.list, successState.callback)
    }

    private fun bindEffect(effectBundle: UserListEffect) {
        when (effectBundle) {
            is NavigateTo -> {
               // ..
            }
        }
    }

    fun bindError(errorMessage: String, cause: Throwable, intent: UserListIntents) {
        //..
    }

    private fun setupRecyclerView() {
        usersAdapter = object : GenericRecyclerViewAdapter() {
            override fun onCreateViewHolder(parent: ViewGroup, viewType: Int): GenericViewHolder<*> {
                return when (viewType) {
                    R.layout.empty_view -> EmptyViewHolder(layoutInflater
                            .inflate(R.layout.empty_view, parent, false))
                    R.layout.user_item_layout -> UserViewHolder(layoutInflater
                            .inflate(R.layout.user_item_layout, parent, false))
                    else -> throw IllegalArgumentException("Could not find view of type $viewType")
                }
            }
        }
        usersAdapter.setAreItemsClickable(true)
        user_list.layoutManager = androidx.recyclerview.widget.LinearLayoutManager(this)
        user_list.adapter = usersAdapter
        usersAdapter.setAllowSelection(true)
        intentStream = flowOf(intentStream, user_list.scrollEvents()
                .map { recyclerViewScrollEvent ->
                    GetPaginatedUsersIntent(
                            if (ScrollEventCalculator.isAtScrollEnd(recyclerViewScrollEvent))
                                viewState.lastId
                            else -1)
                }
                .filter { it.lastId != -1L }
                .conflate()
                .onEach { Log.d("NextPageIntent", "fired!") })
                .flattenMerge()
    }

    fun toggleLoadingViews(isLoading: Boolean, intent: UserListIntents?) {
        linear_layout_loader.bringToFront()
        linear_layout_loader.visibility = if (isLoading) View.VISIBLE else View.GONE
    }
    }

Solution

I think the problem here is intents.consumeEach is waiting for intents to be closed but it is never closed. Which means resultFlows.consumeEach is never reached, so it might look like getUsers is not emitting anything but it's actually not being consumed.

Quick fix would be to wrap each consumeEach in a launch, but I recommend a refactor/re-design.

Side note: Functions that return Flow shouldn't suspend.

Answered By - Dominic Fischer
Answer Checked By - Candace Johnson (JavaFixing Volunteer)