如何使用基于请求参数的Flows在akka-http中引入非阻塞延迟

Question

我正在启动服务器并使用akka流像这样处理它

connection.handleWith(handleRequest())

其中handleRequest():Flow[HttpRequest,HttpRespnse,_]我需要根据查询参数创建一个延迟将响应发送回客户端。我可以提取查询参数，我无法弄清楚，我怎么能用这个来创建延迟。

Answer 1

看看DelayFlow

https://github.com/akka/akka-stream-contrib/blob/master/contrib/src/main/scala/akka/stream/contrib/DelayFlow.scala

https://developer.lightbend.com/docs/api/akka-stream-contrib/current/akka/stream/contrib/DelayFlow.html

它应该允许您在流处理中引入延迟。

Answer 2

扩展@KnowsNotMuch答案，假设您有一些标准来决定是否延迟请求：

val shouldDelayRequest : (HttpRequest) => Boolean = ???

你可以使用这个决策来创建一个DelayStrategy：

import scala.concurrent.duration.{FiniteDuration, TimeUnit}

val noDelay : FiniteDuration = FiniteDuration(0L, TimeUnit.SECONDS)

val createDelayStrategy : (FiniteDuration) => () => DelayStrategy[HttpRequest] = 
  (finiteDelay) => () => new DelayStrategy[HttpRequest] {
    override def nextDelay(elem: HttpRequest) : FiniteDuration = 
      if(shouldDelayRequest(elem))
        finiteDelay
      else
        noDelay
  }

您可以使用此功能创建DelayFlow：

import akka.stream.contrib.DelayFlow

val delay = FiniteDuration(42L, TimeUnit.Seconds)

val delayFlow : DelayFlow[HttpRequest] = DelayFlow(createDelayStrategy(delay))

然后，这个delayFlow可以连接到您处理请求到响应中的任何功能：

val requestToResponseFlow : Flow[HttpRequest, HttpResponse, _] = ???

val possibleDelayedResponseFlow  : Flow[HttpRequest, HttpResponse, _] = 
  delayFlow via requestToResponseFlow