为什么在 Kotlin 中无法启动 emitAll() 之后的 Flow 的 emit() 操作？ onCompletion{} 也无法启动

Question

我运行代码A，如我所料得到结果A，代码A中的所有emit操作都一一启动。

我认为代码B和代码C中的所有emit操作都应该一一启动，但实际上。 emitAll() 之后的 emit() 操作未启动！您可以看到结果 B 和结果 C。为什么？

我运行代码 D 并如我所料得到结果 D，onStart{...} 在 return 流程之前启动。

但是onCompletion{}在代码E和代码F中都没有启动，你可以看到结果E和结果F，为什么？

代码A

@Composable
fun Greeting(
    name: String,
    mViewMode:SoundViewModel= viewModel()
) {
    Column(
        modifier = Modifier.fillMaxSize(),
        verticalArrangement = Arrangement.Center,
        horizontalAlignment = Alignment.CenterHorizontally
    ) {
        LaunchedEffect(Unit) {
            Log.e("My", "Start")

            val s = mViewMode.listRecord()
            s.collect { i ->
                when(i){
                   is  Result.Loading ->  Log.e("My", "Load")
                   is  Result.Success -> Log.e("My", "Success")
                   is  Result.Error ->   Log.e("My", "Error")
                }
            }

        }
    }
}

@HiltViewModel
class SoundViewModel @Inject constructor(
    private val aSoundMeter: RecordRepository  
): ViewModel()
{
    fun listRecord(): Flow<Result<List<MRecord>>> {
        return  aSoundMeter.listRecord()
    }
}


class RecordRepository @Inject constructor(private val mRecordDao:RecordDao): IRecordRepository {
   override fun listRecord(): Flow<Result<List<MRecord>>> {
       return  flow {
           emit(Result.Loading)
           emit(Result.Error(Exception()))
           val s=mRecordDao.listRecord().map { Result.Success(listEntityToModel(it)) }
           emitAll(s)
        }
   }
}


@Dao
interface  RecordDao {   
    @Query("SELECT * FROM record_table ORDER BY createdDate desc")
    fun listRecord():  Flow<List<RecordEntity>>
}


sealed class Result<out R> {
    data class Success<out T>(val data: T) : Result<T>()
    data class Error(val exception: Exception) : Result<Nothing>()
    object Loading : Result<Nothing>()
}

结果A

2022-04-04 18:13:12.417 25866-25866/info.dodata.soundmeter E/My: Start
2022-04-04 18:13:12.428 25866-25866/info.dodata.soundmeter E/My: Load
2022-04-04 18:13:12.429 25866-25866/info.dodata.soundmeter E/My: Error
2022-04-04 18:13:12.509 25866-25866/info.dodata.soundmeter E/My: Success

代码B

//...The same 
 
class RecordRepository @Inject constructor(private val mRecordDao:RecordDao): IRecordRepository {

    override fun listRecord(): Flow<Result<List<MRecord>>> {
       return  flow {
              emit(Result.Loading)
              val s=mRecordDao.listRecord().map { Result.Success(listEntityToModel(it)) }
              emitAll(s)
              emit(Result.Error(Exception()))
        }
    }
}

结果B

2022-04-04 18:15:57.980 26014-26014/info.dodata.soundmeter E/My: Start
2022-04-04 18:15:57.993 26014-26014/info.dodata.soundmeter E/My: Load
2022-04-04 18:15:58.087 26014-26014/info.dodata.soundmeter E/My: Success

代码C

//...The same

class RecordRepository @Inject constructor(private val mRecordDao:RecordDao): IRecordRepository {

    override fun listRecord(): Flow<Result<List<MRecord>>> {
       return  flow {
              val s=mRecordDao.listRecord().map { Result.Success(listEntityToModel(it)) }
              emitAll(s)
              emit(Result.Error(Exception()))
              emit(Result.Loading)
        }
    }
}

结果C

2022-04-04 18:16:57.568 26074-26074/info.dodata.soundmeter E/My: Start
2022-04-04 18:16:57.673 26074-26074/info.dodata.soundmeter E/My: Success

代码D

//...The same 
class RecordRepository @Inject constructor(private val mRecordDao:RecordDao): IRecordRepository {
    override fun listRecord(): Flow<Result<List<MRecord>>> {
       val load: Result<List<MRecord>> =Result.Loading
       val data: Flow<Result<List<MRecord>>> =mRecordDao.listRecord().map { Result.Success(listEntityToModel(it)) }
       return data.onStart { emit(load) }
    }
}

结果D

2022-04-04 18:21:28.409 26170-26170/info.dodata.soundmeter E/My: Start
2022-04-04 18:21:28.423 26170-26170/info.dodata.soundmeter E/My: Load
2022-04-04 18:21:28.482 26170-26170/info.dodata.soundmeter E/My: Success

代码E

//...The same 

class RecordRepository @Inject constructor(private val mRecordDao:RecordDao): IRecordRepository {

    override fun listRecord(): Flow<Result<List<MRecord>>> {
       val load: Result<List<MRecord>> =Result.Loading
       val error: Result<List<MRecord>> =Result.Error(Exception())
       val data: Flow<Result<List<MRecord>>> =mRecordDao.listRecord().map { Result.Success(listEntityToModel(it)) }
       return data.onStart { emit(load) }.onCompletion { emit(error) }
    }

结果E

2022-04-04 18:24:44.053 26272-26272/info.dodata.soundmeter E/My: Start
2022-04-04 18:24:44.067 26272-26272/info.dodata.soundmeter E/My: Load
2022-04-04 18:24:44.142 26272-26272/info.dodata.soundmeter E/My: Success

代码F

//...The same
class RecordRepository @Inject constructor(private val mRecordDao:RecordDao): IRecordRepository {

    override fun listRecord(): Flow<Result<List<MRecord>>> {     
       val error: Result<List<MRecord>> =Result.Error(Exception())
       val data: Flow<Result<List<MRecord>>> =mRecordDao.listRecord().map { Result.Success(listEntityToModel(it)) }
       return data.onCompletion { emit(error) }
    }

}

结果F

2022-04-04 18:25:47.395 26340-26340/info.dodata.soundmeter E/My: Start
2022-04-04 18:25:47.459 26340-26340/info.dodata.soundmeter E/My: Success

Answer 1

来自 Room DAO 的流程永远不会完成。它是无限的，因为它永远监视数据库的变化。因此，当您在其上调用 collect 时（这是 emitAll 在内部所做的），该函数永远不会 returns 并且永远不会访问在其下编写的任何代码。只有当正在收集它的协程被取消时，流程才会停止。

为什么在 Kotlin 中无法启动 emitAll() 之后的 Flow 的 emit() 操作？ onCompletion{} 也无法启动

Why can't the emit() operation of Flow after emitAll() be launched in Kotlin ? and onCompletion{} can't be launched too

kotlin

kotlin-coroutines

kotlin-flow