使用 rest api 进行 Akka Actor 热交换

Question

在调用休息 api 时，我想将演员的路由切换到另一个路由。 请看下面的代码。

几个问题：

1. 代码编译正常，但是当应用程序启动并进行 http 调用时，我得到配置的注册超时 1 秒过期，停止消息，我没有从服务器得到任何响应。
2. 我希望能够通过 api 将路由切换到另一组路由。

制造

package com.example
import akka.actor.Actor
import akka.io.IO
import spray.httpx.RequestBuilding._
import spray.http.MediaTypes._
import spray.routing.{RoutingSettings, RejectionHandler, ExceptionHandler, HttpService}
import spray.util.LoggingContext
import scala.concurrent.Future
import spray.can.Http
import spray.http._
import akka.util.Timeout
import HttpMethods._
import akka.pattern.ask
import akka.event.Logging
import scala.concurrent.duration._

case object Swap
class MyServiceActor extends Actor with MyService with akka.actor.ActorLogging {

  implicit def actorRefFactory = context
  import context._

  def receive = {
      case Swap =>
           become {
             case Swap => unbecome()
             case _    => runRoute(otherRoutes)
           }
      case _ =>   runRoute(myRoute)
  } 
}


trait MyService extends HttpService { this: MyServiceActor =>

  implicit val timeout: Timeout = Timeout(15.seconds)

  implicit val system = context.system

  val myRoute =
  {
    path("") {
      get {
          complete("MyRoute")
      }
    } ~ path("swap") {

        get{
            self ! Swap
            complete("Swapped")
        }
    }
  }

  val otherRoutes =path("") {
  get {
      complete("OtherRoutes")
     }
   } ~ path("swap") {
        get{
        self ! Swap
        complete("Reverted")
     }
}

}

Answer 1

runRoute 是一个部分应用的函数，所以你不能只写runRoute(routeName) 来调用它——它只会返回另一个函数（处理路由）但不调用它；你应该明确地传递请求对象：

def receive = {
      case Swap =>
           become {
             case Swap => unbecome()
             case x    => val f = runRoute(otherRoutes); f(x)
           }
      case x => val f = runRoute(myRoute); f(x)
  } 

runRoute(route) 返回处理“已连接”消息的函数。这就是为什么你会收到“注册超时”错误的原因——你没有从接收方法返回这个函数。当您编写def receive = runRoute(route) 时，此函数用作处理程序，一切都很好。但是当您编写 def receive = {case _ => runRoute(route)} 时，什么也没有发生 - receive 函数什么也不做，因为 runRoute(route) 返回的函数无处可去。

见，https://github.com/spray/spray/blob/master/spray-routing/src/main/scala/spray/routing/HttpService.scala

而且您也可以直接从您的路线调用 become/unbecome，因为您已经将 MyServiceActor 作为 self-type。当您使用单独的Swap 消息时 - 在您收到成功的“交换”响应后，actor 可能会稍微更改其角色（角色更改将异步发生）

case object Swap
class MyServiceActor extends Actor with MyService with akka.actor.ActorLogging {

  implicit def actorRefFactory = context
  import context._

  def swapped = {
      case x => val f = runRoute(otherRoutes); f(x)
  }

  def receive = {
      case x => val f = runRoute(myRoute); f(x)
  } 
}


trait MyService extends HttpService { this: MyServiceActor =>

  implicit val timeout: Timeout = Timeout(15.seconds)

  implicit val system = context.system

  val myRoute = {
    pathSingleSlash {
      get {
            complete("MyRoute")
      }
    } ~ path("swap") {
        get {
            become(swapped)
            complete("Swapped")
        }
    }
  }

  val otherRoutes = { 
   pathSingleSlash {
     get {
      complete("OtherRoutes")
     }
   } ~ path("swap") {
     get{
        unbecome()
        complete("Reverted")
     }
   }
  }
}

更新：您的路径匹配器也不正确。使用：

 pathSingleSlash {
   ...
 } ~ path("swap") {
     ...
 }

或

 path("swap") {
     ...
 } ~ path("") { //matches everything else
     ...
 } 

更新2：

确保您的演员在您的 Main 中注册为单例：

import akka.io.IO
import spray.can.Http

implicit val system = ActorSystem()

val myListener: ActorRef = Props[MyServiceActor]

IO(Http) ! Http.Bind(myListener, interface = "localhost", port = 8080)

http://spray.io/documentation/1.1-SNAPSHOT/spray-can/http-server/#http-server

Answer 2

我也遇到同样的问题，那你可以把服务器参数调大一点来解决这个问题

在此处输入代码

val backlog = 50000
val standardConfig = ServerSettings("spray.can.server.ssl-encryption = off, spray.can.server.registration-timeout = 5s")
val config = standardConfig.copy(serverHeader = "REST API", sslEncryption = false, remoteAddressHeader = true)
val serviceActor = system.actorOf(Props[ApiServiceActor], "apiServiceActor")
IO(Http) ? Http.Bind(serviceActor, interface = "0.0.0.0", ConfigurationHelper.Port, settings = Some(config), backlog = backlog)