callbackFlow ничего не возвращает android kotlin

Question

Я хочу реализовать базу данных в реальном времени firebase с сопрограммами, поэтому мне нужно использовать поток, потому что firebase просто принимает обратные вызовы. проблема в том, что блок .collect {} никогда не запускается, вот мой код

   @ExperimentalCoroutinesApi
override suspend fun getProduct(barcode: String): ProductItem? {
    return withContext(Dispatchers.Default) {
        println("Hi")
        var item: ProductItem? = null
        productFlow(barcode).collect {
        //this never gets called
            print("Getting product")
            item = it
        }
        println("Ending product request ${item?.name}")
        Log.i("GetProduct",item?.name)
        item
    }
}
@ExperimentalCoroutinesApi
private fun productFlow(barcode: String): Flow<ProductItem?> = callbackFlow {
    val database = FirebaseDatabase.getInstance()
    val productRef = database.getReference("products/$barcode")
    val callback = object : ValueEventListener {
        override fun onDataChange(dataSnapshot: DataSnapshot) {
            for(snapshot in dataSnapshot.children){

                Log.i("Source", snapshot.value.toString())
            }
            val product = dataSnapshot.getValue(ProductItem::class.java)
            Log.i("Source",product?.name) //everything is good until here

            sendBlocking(dataSnapshot.getValue(ProductItem::class.java)) //after this i dont get anything on the collect{} block
        }

        override fun onCancelled(databaseError: DatabaseError) {
            println("cancelling")
            sendBlocking(null)
        }
    }
    try {
        productRef.addListenerForSingleValueEvent(callback)
    } catch (e: FirebaseException) {
        println("Firebase exception")
        sendBlocking(null)
    }
    awaitClose{
        println("Closing")
        productRef.removeEventListener(callback)
    }
}

Answer 1

Сначала я бы предложил использовать метод catch, чтобы проверить, есть ли ошибка или нет. Во-вторых, для callbackflow я помню, что использовал offer() вместо sendBlocking