Akka Stream 和 HTTP Scala：如何从路由向 Actor 发送消息答案

【问题标题】：Akka Stream and HTTP Scala: How to send Messages to an Actor from a RouteAkka Stream 和 HTTP Scala：如何从路由向 Actor 发送消息
【发布时间】：2015-12-09 04:39:42
【问题描述】：

我正在玩akka-stream-and-http-experimental 1.0。到目前为止，我有一个可以接受和响应 HTTP 请求的用户服务。我还将有一个可以管理约会的约会服务。为了进行约会，必须是现有用户。如果用户存在，约会服务将与用户服务进行检查。现在这显然可以通过 HTTP 完成，但我宁愿让约会服务向用户服务发送消息。作为新手，我不清楚如何使用演员（如akka-http 抽象）来发送和接收消息。在doc 中提到了ActorRef 和ActorPublisher，但没有前者的例子，而后者对于我的需要来说似乎有点过头了。我的代码如下所示，位于Github:

trait UserReadResource extends ActorPlumbing {
  val userService: UserService
  val readRoute = {
  // route stuff
  }
}

trait ActorPlumbing {
  implicit val system: ActorSystem
  implicit def executor: ExecutionContextExecutor
  implicit val materializer: Materializer

  def config: Config
  val logger: LoggingAdapter
}

trait UserService { // Implemented by Slick and MongoDB in the backend
  def findByFirstName(firstName: String): Future[immutable.Seq[User]]
}

object UserApp extends App with UserReadResource with UserWriteResource with ActorPlumbing {
  override implicit val system = ActorSystem()
  override implicit def executor = system.dispatcher
  override implicit val materializer = ActorMaterializer()

  override def config = ConfigFactory.load()
  override val logger = Logging(system, getClass)

  private val collection = newCollection("users")
  val userRepository = new MongoDBUserRepository(collection)
  val userService: UserService = new MongoDBUserRepositoryAdapter(userRepository) with UserBusinessDelegate { 
    // implicitly finds the executor in scope. Ain't that cute?
    override implicit def executor = implicitly
  } 

  Http().bindAndHandle(readRoute ~ writeRoute, config.getString("http.interface"), config.getInt("http.port"))
}

编辑：我想出了如何发送消息，这可以使用Source.actorRef 完成。那只会将消息发送到流中。我想做的是让路由处理程序类接收响应。这样，当我创建约会服务时，它的参与者可以调用用户服务参与者并以与我示例中的用户路由处理程序相同的方式接收响应。伪代码：

val src = Source.single(name) \\ How to send this to an actor and get the response

编辑 2：根据@yardena 的回答，我想出了以下内容，但最后一行没有编译。我的演员发布者返回一个Future，我猜它会被包装在Promise 中，然后作为Future 传递给路由处理程序。

get {
  parameters("firstName".?, "lastName".?).as(FindByNameRequest) { name =>
    type FindResponse = Future[FindByNameResponse]

    val src: Source[FindResponse, Unit] = Source.actorPublisher[FindResponse](businessDelegateProps).mapMaterializedValue {
      _ ! name
    }
    val emptyResponse = Future.apply(FindByNameResponse(OK, Seq.empty))

    val sink = Sink.fold(emptyResponse)((_, response: FindResponse) => response)

    complete(src.runWith(sink)) // doesn't compile
  }
}

【问题讨论】：

将您的用户服务包装在一个演员中，并通过向该演员发送消息来进行服务调用。然后，您的约会服务（包装在另一个演员中）也可以通过向包装它的用户演员发送相同的消息来与用户服务交互。
@Shadowlands 这就是我打算做的。问题是我如何从路由实现并向参与者发送消息？你能展示在路由中使用ActorRef 的代码示例吗？
不确定您在哪个方面苦苦挣扎 - 您对 Akka 的其余部分有多熟悉？您可以简单地调用一个演员 - 并获得Future 类型的响应 - 使用类似myActorRef ? myMessage 的行，其中消息可以是任何不可变类型（所以从String 到专门构建的案例类，例如case class UserByFirstNameRequest(firstName: String)。例如，参见here。
@Shadowlands 你说的很明显。我想了解的是我的代码中的UserReadResource 如何创建Source[T, ActorRef]，然后将其具体化为Actor。看起来我必须使用Source.actorPublisher(Props) 创建一个发布者，其中的道具将由UserApp 提供。

标签： scala actor akka-stream akka-http

【解决方案1】：

此链接可能会有所帮助：http://zuchos.com/blog/2015/05/23/how-to-write-a-subscriber-for-akka-streams/ 和 @Noah 的回答 Accessing the underlying ActorRef of an akka stream Source created by Source.actorRef

基本上你有两个选择： 1) 如果你想要一个“简单”的actor，它将接收到的所有消息转发到流中，你可以使用 Source.actorRef。然后，您可以通过使用 mapAsync 创建处理阶段将消息通过管道传输到 UserService。 2) 另一种选择，如果您希望演员有一些自定义行为，是编写您自己的 ActorPublisher。

HTH

【讨论】：

快速浏览一下，您的第一个链接似乎是相关的，它也在做我想做的事情。我去看看，然后告诉你。

【解决方案2】：

我最终使用了Actor.ask。很简单。

【讨论】：