如何通过Monix阻止发送HTTP get请求

Question

基于我之前的问题，并根据Artem的见解，我的目标是将get请求发送到给定的url，并使用Monix的节流功能来分隔请求(以避免达到速率限制(。

预期的工作流程如下所示：

make 1 (or more) api call(s) -> apply back-pressure/pausing (based on throttle) -> make the next request -> so on and so forth..

这是我迄今为止所尝试的(下面是我实际代码的简化片段(：

import sttp.client.asynchttpclient.monix._
import monix.eval.Task
import monix.reactive.Observable
import sttp.client.{Response, UriContext}
import scala.concurrent.duration.DurationInt
object ObservableTest extends App {
val activities = AsyncHttpClientMonixBackend().flatMap { implicit backend =>
val ids: Task[List[Int]] = Task { (1 to 3).toList }
val f: String => Task[Response[Either[String, String]]] = (i: String) => fetch(uri"$i", "")
val data: Task[List[Task[Response[Either[String, String]]]]] = ids map (_ map (_ => f("https://heloooo.free.beeceptor.com/my/api/path")))
data.guarantee(backend.close())
}
import monix.execution.Scheduler.Implicits.global
val flat: Unit = activities.runToFuture.foreach { x =>
val r: List[Task[Response[Either[String, String]]]] = x // List with size 3
Observable
.fromIterable(r)
.throttle(6 second, 1)
.map(_.runToFuture)
.subscribe()
}
while (true) {}
}

这就是获取数据的函数的样子：

def fetch(uri: Uri, auth: String)(implicit
backend: SttpBackend[Task, Observable[ByteBuffer], WebSocketHandler]
) = {
println(uri)
val task = basicRequest
.get(uri)
.header("accept", "application/json")
.header("Authorization", auth)
.response(asString)
.send()
task
}

我已经尝试过运行前面提到的代码，但我仍然看到所有的get请求都是在没有任何间隔的情况下发出的。

举例来说，我当前的api调用日志如下所示：

//(https://mr.foos.api/v1), Sat Aug 08 18:47:15 CEST 2020)
//(https://mr.foos.api/v1), Sat Aug 08 18:47:15 CEST 2020)
//(https://mr.foos.api/v1), Sat Aug 08 18:47:15 CEST 2020)
//(https://mr.foos.api/v1), Sat Aug 08 18:47:15 CEST 2020)

我正在努力实现类似的目标：

//(https://mr.foos.api/v1), Sat Aug 08 18:50:15 CEST 2020)
//(https://mr.foos.api/v1), Sat Aug 08 18:50:18 CEST 2020)
//(https://mr.foos.api/v1), Sat Aug 08 18:50:21 CEST 2020)
//(https://mr.foos.api/v1), Sat Aug 08 18:50:24 CEST 2020)

更新：

• 我已经使用Beeceptor将api设置为可模拟的。在我看来，print语句是由调用函数生成的，但实际上并没有发送请求。我还更新了我的函数调用，将其解析为字符串(只是为了简单起见(。然而，当我试图将请求限制到mock api时，它仍然没有收到任何请求

Answer 1

所以如果我理解得对，你就有这样的类型：

object ObservableTest extends App  {
type Response = Either[ResponseError[Error], Activities]
case class Activities()
val activities: Task[List[Response]] = AsyncHttpClientMonixBackend().flatMap { implicit backend =>
def fetchData(uri: String): Task[Response] = ???
val ids: Task[List[Int]] = ??? // a Task containing a List of IDs (from a previous step)
val func: String => Task[Response] = (i: String) => fetchData("someUri") // a function that will be used to call an endpoint
val data: Task[List[Task[Response]]] = ids map (_ map (id => func(id.toString))) // Maps API calling-function to the ids
val activitiesData: Task[List[Response]] = data.flatMap(Task.parSequenceUnordered(_)) // Flattenned the previous step
activitiesData.guarantee(backend.close())
}
import monix.execution.Scheduler.Implicits.global
Observable(activities)
.throttle(3 second, 1)
.subscribe()
}

代码中的问题是，您要限制一个包含多个操作的大任务，其中一些操作甚至是并行的(但这不是问题的根源(。即使在类型中，你也可以看到——你应该从任务列表中进行观察(每个任务都会被抑制(，而不是从列表中的任务中进行观察。

事实上，我不知道id来自哪里，它可能是评估管道的基石。但是，如果我们像示例中那样与他们一起执行任务。我们会这样做的。

import monix.eval.Task
import sttp.client.asynchttpclient.monix._
import monix.eval.Task._
import monix.reactive.Observable
import sttp.client.ResponseError
import scala.concurrent.duration.DurationInt
object ObservableTest extends App  {
type Response = Either[ResponseError[Error], Activity]
case class Activity()
val activities: Task[List[Task[Response]]] = AsyncHttpClientMonixBackend().flatMap { implicit backend =>
def fetchData(uri: String): Task[Response] = Task {
println("mocked http request")
Right(Activity())
}
val ids: Task[List[Int]] = Task { (1 to 100).toList} // a Task containing a List of IDs (from a previous step)
val func: Int => Task[Response] = (i: Int) => fetchData(s"someUri_$i") // a function that will be used to call an endpoint
val data: Task[List[Task[Response]]] = ids.map(_.map(func)) // Maps API calling-function to the ids
data.guarantee(backend.close())
}
import monix.execution.Scheduler.Implicits.global
Observable.fromTask(activities)
.flatMap { listOfFetches: List[Task[Response]]  =>
Observable.fromIterable(listOfFetches)
}
.throttle(3.second, 1)
.map(_.runToFuture) 
.subscribe()

while(true) {}
}

我们限制一个获取列表，而不是在其中执行所有获取的任务。

PS：请问一些不清楚的问题，我会在代码中添加评论