标签: kotlin-flow

当用户快速点击按钮时，showDialog() 方法会在彼此的顶部显示多次，因此当您关闭它时，它后面还有另一个。我正在寻找一种方法来忽略第二次点击 1 秒而不使用处理程序或检查前一次点击的时间。

//Button that opens a dialog
button.setOnClickListener {
    showDialog()
}

我正在寻找使用 Kotlin 协程或 Kotlin 流程的解决方案以供将来实现。

我有以下代码：

val channel = BroadcastChannel<Event>(10)

fun setup() {
    scope.launch {
        channel.asFlow().
            .flatMapLatest { fetchSomeData() }
            .catch { emit(DefaultData()) }
            .onEach { handleData() }
            .collect()

    }
}

fun load() {
    channel.offer(Event.Load)      
}

如果fetchSomeData因异常而失败，它将被捕获catch并传递一些默认数据。问题是流本身被取消并从频道的订阅者中删除。这意味着提供给频道的任何新事件都将被忽略，因为不再有任何订阅者。

有没有办法确保在发生异常时流不会被取消？

我正在尝试使用Kotlin Flows. 这是我现在正在尝试的

flowOf(
 remoteDataSource.getDataFromCache() // suspending function returning Flow<Data>
   .catch { error -> Timber.e(error) },
 remoteDataSource.getDataFromServer() // suspending function returning Flow<Data>
).flattenConcat().collect {
  Timber.i("Response Received")
}

这里的问题collect是只在getDataFromServer返回时调用。我的期望是我应该在几毫秒后从缓存中获取第一个事件，然后从服务器获取第二个事件。在这种情况下，"Response Received"会打印两次，但会立即一个接一个地打印。

在此其他变体中，"Response Received"仅在getDataFromServer()返回后打印一次。

 remoteDataSource.getDataFromCache() // suspending function returning Flow<Data>
  .catch { error -> Timber.e(error) }
  .flatMapConcat {
    remoteDataSource.getDataFromServer() // suspending function returning Flow<Data>
  }
  .collect {
    Timber.i("Response Received")
  }

我Flowable.concat()之前使用过 RxJava ，它运行良好。Kotlin Flows 中是否有可以模拟这种行为的东西？

我有一个List<Flow<T>>，想生成一个Flow<List<T>>. 这几乎是什么combine- 除了组合等待每个都Flow发出初始值，这不是我想要的。以这段代码为例：

val a = flow {
  repeat(3) {
    emit("a$it")
    delay(100)
  }
}
val b = flow {
  repeat(3) {
    delay(150)
    emit("b$it")
  }
}
val c = flow {
  delay(400)
  emit("c")
}
val flows = listOf(a, b, c)
runBlocking {
  combine(flows) {
    it.toList()
  }.collect { println(it) }
}

使用combine（因此按原样），这是输出：

[a2, b1, c]
[a2, b2, c]

而我也对所有中间步骤感兴趣。这就是我想要从这三个流程中得到的：

[]
[a0]
[a1]
[a1, b0]
[a2, b0]
[a2, b1]
[a2, b1, c]
[a2, …

我无法理解catch运算符在 kotlin 中的工作原理Flow。 这是捕获文档

问题：

1. 为什么 a 的存在不允许catch在Flow遇到异常时继续，而不是完成？
2. 操作员的放置catch似乎改变了行为。为什么我不能将catch运算符放在链的末尾以看到相同的结果？在我的示例中，只有当我将其放置在 BEFORE 之前，它才会执行onEach。

示例要点

第一个例子，放置catchBEFORE onEach：

fun main() {
    // Flow of lambdas that return a String (or throw an Exception)
    flowOf<() -> String>({ "Hello " }, { error("error") }, { "World" })
        // Map to the result of the invocation of the lambda
        .map { it() }
        // This line will emit the error …

我的 LazyColumn 没有重组，但值正在更新。

如果我向下滚动列表并向上滚动，我会看到 UI 的正确值

主要活动

class MainActivity : AppCompatActivity() {


    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContent {
            MyTheme {
                MyApp()
            }
        }
    }
}

// Start building your app here!
@Composable
fun MyApp(vm: PuppyListViewModel =  viewModel()) {
    val puppers by vm.pups.collectAsState(emptyList())
    Surface(color = MaterialTheme.colors.background) {
        Column {
            Toolbar()
            LazyColumn {
                items(puppers) { pup ->  PuppyUI(pup, vm::seeDetails, vm::togglePuppyAdoption) }
            }
        }
    }
}

视图模型

class PuppyListViewModel : ViewModel() {

    val pups = PuppyRepo.getPuppies().onEach {
        println("FlowEmitted: $it")
    }

    fun …

当导航回片段时，如何避免再次执行collect{}代码。

视图模型类

    private val _commitResult = MutableStateFlow<Map<String, Any>>(mapOf())
    val commitResult: StateFlow<Map<String, Any>> = _commitResult
Fragment code like this:

    viewLifecycleOwner.lifecycleScope.launch {
            viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED){
                viewModel.commitResult.collect { data ->
                    Logger.i("commitResult $data")
                    //navigate to another fragment
                }
            }
        }

当我首先更改 viewModel 中的 _commitResult 值时，跳转到另一个片段效果很好。不幸的是，当我回到片段时。collect{ // navigate to another fragment}将再次执行。

我知道什么时候回到片段。onCreateView 再次执行，viewModel 会发出之前的数据存储，因此执行collect { // navigate to another fragment}。我怎样才能避免这种情况？

与 LiveData 相同，我使用 Event 来修复 LiveData 的问题。

open class Event<out T>(private val content: T) {

var hasBeenHandled = false
    private set // Allow …

我遇到一种情况，我需要观察 userId，然后使用这些 userId 来观察用户。userIds 或用户可能随时更改，我希望使发出的用户保持最新状态。这是我拥有的数据源的示例：

data class User(val name: String)

fun observeBestUserIds(): Flow<List<String>> {
    return flow {
        emit(listOf("abc", "def"))
        delay(500)
        emit(listOf("123", "234"))
    }
}

fun observeUserForId(userId: String): Flow<User> {
    return flow {
        emit(User("${userId}_name"))
        delay(2000)
        emit(User("${userId}_name_updated"))
    }
}

在这种情况下，我希望排放量为：

[User(abc_name), User(def_name)]， 然后

[User(123_name), User(234_name)]， 然后

[User(123_name_updated), User(234_name_updated)]

我想我可以在 RxJava 中实现这一点，如下所示：

observeBestUserIds.concatMapSingle { ids ->
    Observable.fromIterable(ids)
        .concatMap { id ->
            observeUserForId(id)
        }
        .toList()
}

我应该编写什么函数来生成发出该信号的流？

当我们平时使用 Room 时，我们使用 Kotlin Coroutine 并创建一个 DAO 来访问 Room 并获取结果。大多数函数通常suspend在函数开头有修饰符 butLiveData和Flow。例如，让我们看一下下面的这两段代码。

@Query("SELECT * FROM MockTable")
suspend fun allMockDataWithSuspend(): List<MockData>

@Query("SELECT * FROM MockTable")
fun allMockData(): Flow<List<MockData>> // or LiveData<List<MockData>>

当我们使用suspend修饰符时，我们需要在协程范围内调用该函数，因为该函数具有 suspend 修饰符。LiveData但是当函数的结果是或者Flow即使是 I/O 访问时，我们不需要在协程中调用该函数。

这怎么可能？

我有一个可能永远不会被调用的监听器。但是，如果它至少被调用一次，我有理由确信它会被调用很多次。我是 Flows 的粉丝，因此我将其包装在callbackFlow() 构建器中。为了防止永远等待，我想添加一个超时时间。我正在尝试构建流运算符，如果流的第一个元素发出时间太长，它将抛出某种超时。这是我所拥有的。

fun <T> Flow<T>.flowBeforeTimeout(ms: Long): Flow<T> = flow {
    withTimeout(ms){
        emit(first())
    }
    emitAll(this@flowBeforeTimeout.drop(1))
}

效果还不错，这些 JUnit4 测试都通过了。还有更多通过的测试，但为了简洁起见我省略了它们。

@Test(expected = CancellationException::class)
fun `Throws on timeout`(): Unit = runBlocking {
    val testFlow = flow {
        delay(200)
        emit(1)
    }

    testFlow.flowBeforeTimeout(100).toList()
}

@Test
fun `No distortion`(): Unit = runBlocking {
    val testList = listOf(1,2,3)

    val resultList = testList
        .asFlow()
        .flowBeforeTimeout(100)
        .toList()

    assertThat(testList.size, `is`(resultList.size))
}

然而，这个测试没有通过。

// Fails with: Expected: is <1> but: was <2>
@Test
fun `Starts only once`(): Unit = …