回调流等待收集其内部流

Question

我正在我的Android应用程序中实现Firestore，我想知道如何使该代码在使用该程序之前等待其内部流程。 Trip.tripdays在wher表达式中的响应之后的时间填充了数据。

override fun getTrips(idUser: String): Flow<Response<List<Trip>>> = callbackFlow {
        val trips = mutableListOf<Trip>()
        tripsRef
            .whereEqualTo(ID_USER, idUser)
            .get()
            .addOnSuccessListener { snapshot ->
                for (doc in snapshot.documents) {
                    val trip = doc.toObject(Trip::class.java)
                    if (trip != null) {
                        trip.id = doc.id
                        CoroutineScope(Dispatchers.IO).launch {
                            val tripDaysFlow = tripDaysRepository.getTripDays(trip.id)
                            tripDaysFlow.collect() {
                                when (it) {
                                    is Response.Loading -> {
                                        Log.d("FirestoreTripsRepositoryImpl", "Loading")
                                    }
                                    is Response.Success -> {
                                        trip.tripDays = it.data
                                    }
                                    is Response.Error -> {
                                        trySend(Response.Error(it.message)).isFailure
                                    }
                                    is Response.Message -> {
                                        trySend(Response.Message(it.message))
                                    }
                                }
                            }
                        }
                        trips.add(trip)
                    }
                }
                trySend(Response.Success(trips)).isSuccess
            }
            .addOnFailureListener {
                trySend(Response.Error(it.message ?: it.toString())).isFailure
            }
            .await()
        awaitClose()
    }

override fun getTripDays(idTrip: String): Flow<Response<List<TripDay>>> = callbackFlow {
        val tripDaysRef = tripsRef.document(idTrip).collection(TRIP_DAYS_REF)
        val tripDays = mutableListOf<TripDay>()
        tripDaysRef
            .get()
            .addOnSuccessListener {
                for (doc in it.documents) {
                    val tripDay = doc.toObject(TripDay::class.java)
                    if (tripDay != null) {
                        tripDay.id = doc.id
                        tripDays.add(tripDay)
                    }
                }
                trySend(Response.Success(tripDays)).isSuccess
            }
            .addOnFailureListener {
                trySend(Response.Error(it.message ?: it.toString())).isFailure
            }
            .await()
        awaitClose ()
    }

sealed class Response<out T> {
    object Loading : Response<Nothing>()
    data class Success<out T>(val data: T) : Response<T>()
    data class Error(val message: String) : Response<Nothing>()
    data class Message(val message: String) : Response<Nothing>()
}

一个很大的发射：

override fun getTrips(idUser: String): Flow<Response<List<Trip>>> = callbackFlow {
        val trips = mutableListOf<Trip>()
        tripsRef
            .whereEqualTo(ID_USER, idUser)
            .get()
            .addOnSuccessListener { snapshot ->
                launch {
                    Log.d("launch", "launched")
                    for (doc in snapshot.documents) {
                        if (doc.id == "kJJFEatove5mxw4P70uq")
                            continue
                        val trip = doc.toObject(Trip::class.java)
                        if (trip != null) {
                            trip.id = doc.id
                            val tripDaysFlow = tripDaysRepository.getTripDays(trip.id)
                            tripDaysFlow.collect {
                                when (it) {
                                    is Response.Loading -> trySend(Response.Loading)
                                    is Response.Success -> trip.tripDays = it.data
                                    is Response.Error -> trySend(Response.Error(it.message)).isFailure
                                    is Response.Message -> trySend(Response.Message(it.message))
                                }
                            }
                            trips.add(trip)
                        }
                    }
                    trySend(Response.Success(trips)).isSuccess
                    close()
                }
            }
            .addOnFailureListener {
                trySend(Response.Error(it.message ?: it.toString())).isFailure
                close()
            }
        awaitClose()
    }

可以使它在没有完整的重建的情况下使其正常工作会很酷，但是让我知道我的方法是否完全错误。干杯!