如何在不升级到Akka HTTP的情况下停止使用路由DSL的Spray服务器？

Question

我有这条路线:

val route = pathPrefix("es") {
  path("se") {
    post {
      entity(as[JsValue]) {
        t =>
          complete("ok")
      }
    }
  } ~ path("q" / "show") {
    get {
      complete(q)
    }
  }
}

当我尝试绑定它以阻止它时(根据https://doc.akka.io/docs/akka-http/current/routing-dsl/index.html),我收到编译错误:

val bindingFuture = Http().bindAndHandle(route, "0.0.0.0", 9100)

错误:(54,46)类型不匹配; 发现:spray.routing.Route(扩展为)spray.routing.RequestContext =>需要的单位:akka.stream.scaladsl.Flow [akka.http.scaladsl.model.HttpRequest,akka.http.scaladsl.model.HttpResponse,任何] val bindingFuture = Http().bindAndHandle(route,"0.0.0.0",9100)

如何停止HTTP服务器？目前我可以使用以下命令启动HTTP服务器:

startServer("0.0.0.0", port)

但是,我没有看到如何使用该startServer功能来阻止它.

更新:我无法按照下面的建议从Spray升级到Akka HTTP(管理,不在我的控制范围内).

看Http().bindAndHandle,它来自akka-http-core_2.11-2.4.11.1.jar.我在这里看到我需要RouteResult将其转换为Flow.但是,我找不到任何RouteResult在akka-http-core_2.11-2.4.11.1.jar.

Answer 1

正如其他答案已经表明的那样,你正在混淆Spray和Akka HTTP.这两个库是不同的,它们各自的服务器端组件并不意味着在同一个应用程序中共存.如果您无法迁移到取代Spray 的Akka HTTP,则从项目中删除Akka HTTP依赖项,并查看Spray文档以获取有关停止Spray服务器的信息:

要显式停止服务器,请Http.Unbind向HttpListener实例发送命令(ActorRef此实例可用作Http.Bound启动服务器时确认事件的发送方).

Http.Unbound在成功解除端口绑定后(或者Http.CommandFailed在出错的情况下),侦听器将回复事件.此时服务器不会接受进一步的请求.

显然你正在使用SimpleRoutingApp,这startServer是定义方法的地方.此方法不公开获取对HttpListeneractor 的引用的方法.如引用文档所述,您必须Http.Unbind向此actor 发送消息才能停止服务器.

一个想法是定义自己的actor,可以发送对以下内容的引用HttpListener:

import akka.actor._
import spray.can.Http
import spray.routing._

object MyActor {
  case object GetListener
  def props(route: => Route): Props = Props(new MyActor(route))
}

class MyActor(route: => Route) extends HttpServiceActor {
  import MyActor._

  var httpListener: Option[ActorRef] = None

  def routeReceive: Receive = runRoute(route)

  def serverLifecycleReceive: Receive = {
    case b: Http.Bound =>
      println(s"Successfully bound to ${b.localAddress}")
      val listener = sender()
      httpListener = Some(listener)
    case GetListener =>
      httpListener.foreach(sender ! _)
  }

  def receive = routeReceive orElse serverLifecycleReceive
}

然后使用此actor代替SimpleRoutingApp启动服务器:

import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.Success
import akka.actor._
import akka.io.IO
import akka.pattern.ask
import akka.util.Timeout
import spray.can.Http
import spray.http._
import spray.routing._
import MyActor

object Main extends App {

  implicit val system = ActorSystem()
  import system.dispatcher
  implicit val timeout = Timeout(5.seconds)

  val route = ???

  val handler = system.actorOf(MyActor.props(route), name = "handler")

  IO(Http) ! Http.Bind(handler, interface = "0.0.0.0", port = 9100)

  // run the below code to shut down the server before shutting down the actor system
  (handler ? MyActor.GetListener)
    .flatMap { case actor: ActorRef => (actor ? Http.Unbind) }
    .onComplete {
      case Success(u: Http.Unbound) =>
        println("Unbinding from the port is done.")
        // system.shutdown()
      case _ =>
        println("Unbinding failed.")
    }
}

所有这些都假设您希望在关闭actor系统之前(或不关闭)显式关闭服务器.如果不是这种情况,您当然可以在不明确停止服务器的情况下关闭actor系统.例如,您可以添加一个路径来处理这个路径(下面的代码是根据Spray存储库中的一个示例应用程序改编的):

object Main extends App with SimpleRoutingApp {
  implicit val system = ActorSystem("simple-routing-app")
  import system.dispatcher

  val route = ...
    ~ (post | parameter('method ! "post")) {
      path("stop") {
        complete {
          system.scheduler.scheduleOnce(1.second)(system.shutdown())(system.dispatcher)
          "Shutting down in 1 second..."
        }
      }
    }

  startServer("0.0.0.0", port = 9100) {
    route
  }.onComplete {
    case Success(b) =>
      println(s"Successfully bound to ${b.localAddress}")
    case Failure(ex) =>
      println(ex.getMessage)
      system.shutdown()
  }
}