kotlin - 延迟 Kotlin 的 buildSequence 的推荐方法是什么？

Question

我正在尝试轮询分页 API 并在新项目出现时向用户提供它们。

fun connect(): Sequence<T> = buildSequence {
    while (true) {
        // result is a List<T>
        val result = dataSource.getFirstPage()
        yieldAll(/* the new data in `result` */)

        // Block the thread for a little bit
    }
}

这是示例用法:

for (item in connect()) {
    // do something as each item is made available
}

我的第一个想法是使用 delay功能，但我收到此消息:

Restricted suspended functions can only invoke member or extension suspending functions on their restricted coroutine scope

这是 buildSequence 的签名:

public fun <T> buildSequence(builderAction: suspend SequenceBuilder<T>.() -> Unit): Sequence<T>

我认为这条消息意味着我只能使用 suspend SequenceBuilder 中的函数: yield和 yieldAll并且使用任意 suspend不允许函数调用。

现在，我正在使用它在每次轮询 API 后阻止序列构建一秒:

val resumeTime = System.nanoTime() + TimeUnit.SECONDS.toNanos(1)
while (resumeTime > System.nanoTime()) {
    // do nothing
}

这有效，但它似乎真的不是一个好的解决方案。有没有人遇到过这个问题？

Answer 1

为什么它不起作用？有些研究

当我们看buildSequence ，我们可以看到它需要一个 builderAction: suspend SequenceBuilder<T>.() -> Unit作为其论据。作为该方法的客户，您将能够收到 suspend具有 SequenceBuilder 的 lambda作为它的接收者(阅读有关 lambda 与接收者 here )。SequenceBuilder本身用 RestrictSuspension 注释:

@RestrictsSuspension
@SinceKotlin("1.1")
public abstract class SequenceBuilder<in T> ...

注释的定义和注释如下:

/**
 * Classes and interfaces marked with this annotation are restricted
 * when used as receivers for extension `suspend` functions. 
 * These `suspend` extensions can only invoke other member or extension     
 * `suspend` functions on this particular receiver only 
 * and are restricted from calling arbitrary suspension functions.
 */
@SinceKotlin("1.1") @Target(AnnotationTarget.CLASS) @Retention(AnnotationRetention.BINARY)
public annotation class RestrictsSuspension

如 RestrictSuspension文档说明，在 buildSequence 的情况下，您可以使用 SequenceBuilder 传递一个 lambda作为它的接收器，但带有 受限 可能性，因为您只能在此特定接收器上调用“其他成员或分机 suspend 功能”。这意味着，块传递给 buildSequence可以调用定义在 SequenceBuilder 上的任何方法(如 yield 、 yieldAll )。另一方面，由于该块“被限制调用任意挂起函数”，使用 delay不起作用。由此产生的编译器错误验证它:

Restricted suspended functions can only invoke member or extension suspending functions on their restricted coroutine scope.

最终，您需要知道 buildSequence创建一个作为同步协程示例的协程。在您的示例中，序列代码将在通过调用 connect() 使用序列的同一线程中执行。 .

如何延迟序列？

据我们了解， buildSequence创建一个同步序列。在这里使用常规线程阻塞是可以的:

fun connect(): Sequence<T> = buildSequence {
    while (true) {
        val result = dataSource.getFirstPage()
        yieldAll(result)
        Thread.sleep(1000)
    }
}

但是，您真的希望整个线程都被阻塞吗？或者，您可以实现异步序列，如 here 所述.结果，使用 delay其他暂停功能有效。