Akka Actor 热插拔使用 rest api

调用 rest api 时，我想将 actor 的路由切换到另一个路由。请参阅下面的代码。

几个问题：

代码编译良好，但是当应用程序启动并进行http调用时，我收到配置的注册超时为1秒过期，停止消息，并且没有收到来自服务器的任何响应。
我希望能够通过 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")
     }
}
}

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

你也可以从你的路线上调用成为/取消成为，因为你已经有MyServiceActor作为自我类型。当您使用单独的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：

确保您的演员在您的主节点中注册为单例：

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

我也面临同样的问题，那么您可以将服务器参数设置得更大以解决问题

在此处输入代码

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)

相关内容

最新更新

热门标签：