如何在akka http路由中处理长时间运行的请求答案

【问题标题】：how to handle long running request in akka http route如何在akka http路由中处理长时间运行的请求
【发布时间】：2021-07-30 08:19:51
【问题描述】：

我正在使用 akka http 我的一条路线是通过 akka http 客户端 api 与外部服务交互，并且 httpRequest 持续运行我无法使其工作这是我的用例->我正在与 janus 服务器交互，并在服务器以“keepAlive”或“事件”响应时立即执行长轮询获取请求，我再次请求等等服务器继续响应

所有这一切都发生在一个演员内部，我有一个 akka htttp 路由，它正在初始化第一个请求

这是我的代码

final case class CreateLongPollRequest(sessionId:BigInt)
class LongPollRequestActor (config: Config) extends Actor {
  def receive = {
    case CreateLongPollRequest(sessionId) =>
      senderRef = Some(sender())
      val uri: String = "localhost:8080/" + sessionId
      val request = HttpRequest(HttpMethods.GET, uri)
      val responseFuture = Http(context.system).singleRequest(request)
      responseFuture
        .onComplete {
          case Success(res)
          Unmarshal(res.entity.toStrict(40 seconds)).value.map { result =>
              val responseStr = result.data.utf8String
              log.info("Actor LongPollRequestActor: long poll responseStr {}",responseStr)
              senderRef match {
                case Some(ref) =>
                  ref ! responseStr
                case None => log.info("Actor LongPollRequestActor: sender ref is null")
              }
            }
          case Failure(e) =>log.error(e)
          }
          }
          }


final case class JanusLongPollRequest(sessionId: BigInt)
class JanusManagerActor(childMaker: List[ActorRefFactory => ActorRef]) extends Actor {
  var senderRef: Option[akka.actor.ActorRef] = None
  val longPollRequestActor  = childMaker(1)(context)
  def receive: PartialFunction[Any, Unit] = {
    case JanusLongPollRequest(sessionId)=>
      senderRef = Some(sender)
      keepAlive(sessionId,senderRef)
}

  

    def keepAlive(sessionId:BigInt,sender__Ref: Option[ActorRef]):Unit= {
        val senderRef = sender__Ref
        val future = ask(longPollRequestActor, CreateLongPollRequest(sessionId)).mapTo[String] //.pipeTo(sender)
    if (janus.equals("keepalive")) {
                val janusRequestResponse = Future {
                  JanusSessionRequestResponse(janus = janus)
                }
                senderRef match {
                  case Some(sender_ref) =>
                    janusRequestResponse.pipeTo(sender_ref)
                }
                keepAlive(sessionId,senderRef)
              }
              else if (janus.equals("event")) {
               //some fetching of values from server 
                val janusLongPollRequestResponse = Future {
                  JanusLongPollRequestResponse(janus = janus,sender=sender, transaction=transaction,pluginData=Some(pluginData))
                }
                senderRef match {
                  case Some(sender_ref) =>
                    janusLongPollRequestResponse.pipeTo(sender_ref)
                }
                keepAlive(sessionId,senderRef)
              }



def createLongPollRequest: server.Route =
    path("create-long-poll-request") {
      post {
entity(as[JsValue]) {
          json =>
            val sessionID = json.asJsObject.fields("sessionID").convertTo[String]
        
          val future = ask(janusManagerActor, JanusLongPollRequest(sessionID)).mapTo[JanusSessionRequestResponse]
            onComplete(future) {
              case Success(sessionDetails) =>
                    log.info("janus long poll request created")
                    val jsonResponse = JsObject("longpollDetails" -> sessionDetails.toJson)
                    complete(OK, routeResponseMessage.getResponse(StatusCodes.OK.intValue, ServerMessages.JANUS_SESSION_CREATED, jsonResponse))
                
              case Failure(ex) =>
                failWith(ex)
            }
  
           }
          }

现在上述路线createLongPollRequest 第一次运行良好，我第一次可以看到响应，在接下来的尝试中，我收到了如下的死信

[INFO] [akkaDeadLetter][07/30/2021 12:13:53.587] [demo-Janus-ActorSystem-akka.actor.default-dispatcher-6] [akka://demo-Janus-ActorSystem/deadLetters] Message [com.ifkaar.lufz.janus.models.janus.JanusSessionRequestResponse] from Actor[akka://demo-Janus-ActorSystem/user/ActorManager/ManagerActor#-721316187] to Actor[akka://demo-Janus-ActorSystem/deadLetters] was not delivered. [4] dead letters encountered. If this is not an expected behavior then Actor[akka://demo-Janus-ActorSystem/deadLetters] may have terminated unexpectedly. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

这可能是第一次迭代后导致问题的原因 responseFuture.pipeTo(sender()

有没有一种方法可以在我的后端服务器响应时在我的 akkahttp 路由中获得响应？

【问题讨论】：

标签： scala akka akka-http

【解决方案1】：

Actor 应该只回复一次CreateLongPollRequest，并且只有在它有有效数据时才应该这样做。如果轮询失败，Actor 应该发出另一个轮询请求。

如果没有 Actor 的详细信息，很难提供更多帮助。

【讨论】：