标签: akka-http

我并不是说播放禁用PlayNettyServer并启用AkkaHttpServer的方式,使用以下文档中描述的AkkaHttpServer:

lazy val root = (project in file("."))
  .enablePlugins(PlayScala, PlayAkkaHttpServer)
  .disablePlugins(PlayNettyServer)

我的意思是利用play框架的依赖注入和play-slick等其他工具,并直接在代码中使用akka-http:

class AppInitiation @Inject()(implicit val system: ActorSystem, configuration: Configuration) {

  implicit val materializer = ActorMaterializer()

  implicit val timeout: Timeout = 5 seconds

  val logger = Logger("Server")

  val milkywayPath = path(Segment ~ RestPath)
  val methods = get | put | post | delete
  val gatewayExceptionHandler = ExceptionHandler {
    case e: AskTimeoutException =>
      complete(HttpResponse(InternalServerError, Nil, "Ask timeout!"))
    case e: Exception =>
      logger.error("unknown error", e)
      complete(HttpResponse(InternalServerError, Nil, "Unknown error! Please contact administratro!")) …

我的喷雾 json 支持看起来像这样

object MarshallingSupport extends SprayJsonSupport {
  implicit def json4sFormats: Formats = DefaultFormats
}

在我的路线中，我想将请求映射到 dto

object Main extends App with AppConfig with BaseService with MainActorSystem {

  val processor = system.actorOf(Props(), "processorActor")
  val view = system.actorOf(Props(), "processorActor")

  override protected implicit val executor: ExecutionContext = system.dispatcher
  override protected val log: LoggingAdapter = Logging(system, getClass)
  override protected implicit val materializer: ActorMaterializer = ActorMaterializer()

  Http().bindAndHandle(routes(processor, view), httpInterface, httpPort)
}

trait BaseServiceRoute {
  protected implicit def executor: ExecutionContext
  protected implicit def materializer: ActorMaterializer …

听起来像是一个简单的问题，但我找不到答案（可能是因为在搜索“如何使用 Akka-HTTP 执行 HTTP/2.0”时，“2.0”被解释为 Akka 的版本）。

似乎 akka-http 确实有一个用于 HttpProtocols 的构造，但它只有 HTTP 1.0 和 HTTP 1.1。

在我的一个项目中，我有一个 akka 演员，用于向我的 google fcm 服务器发送帖子请求。演员接受一个 id 列表，并且应该发出与列表中一样多的请求。我打印出来自服务器的响应，runForeach(println(_))但我只得到一个完整的 id 列表的打印输出。为什么会发生这种情况？

class FCMActor(val key: String) extends Actor{
  import fcm.FCMActor._
  import akka.pattern.pipe
  import context.dispatcher

  private implicit def system: ActorSystem = ActorSystem()
  final implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(context.system))

  def buildBody(id: Option[String]): String = {
    Json.obj(
      "to" -> id,
      "priority" -> "high",
      "data" -> Json.obj("message" -> "Firebase Clud Message"),
      "time_to_live" -> 60
    ).toString()
  }

  def buildHttpRequest(body: String): HttpRequest = {
    HttpRequest(method = HttpMethods.POST,
      uri = s"/fcm/send",
      entity = HttpEntity(MediaTypes.`application/json`, body),
      headers …

如何从 Akka HTTP 路由向 Akka Sink 发送元素/消息？我的 HTTP 路由仍然需要返回一个正常的 HTTP 响应。

我想这需要一个流分支/连接点。正常的 HTTP 路由是来自 HttpRequest -> HttpResponse 的流。我想添加一个分支/连接点，以便 HttpRequests 可以将事件触发到我的单独接收器并生成正常的 HttpResponse。

下面是一个非常简单的单路由 akka-http 应用程序。为简单起见，我使用了一个简单的 println 接收器。我的生产用例，显然会涉及一个不那么琐碎的接收器。

def main(args: Array[String]): Unit = {
  implicit val actorSystem = ActorSystem("my-akka-http-test")
  val executor = actorSystem.dispatcher
  implicit val materializer = ActorMaterializer()(actorSystem)

  // I would like to send elements to this sink in response to HTTP GET operations.
  val sink: Sink[Any, Future[Done]] = Sink.foreach(println)

  val route: akka.http.scaladsl.server.Route =
    path("hello" / Segment) { p =>
      get { …

我正进入（状态

< HTTP/1.1 400 Bad Request
< Content-Type: text/plain; charset=UTF-8
< Content-Length: 96

Illegal request-target: Invalid input '\', expected pchar, '/', '?' or 'EOI' (line 1, column 17)

对于包含 的 URL \，这没关系，我希望有 400 个，但我想更改消息，因此它对用户更友好。

似乎它发生在它进入任何控制器之前。

ps 我知道有akka.http.parsing.uri-parsing-mode = relaxed，但我不想这样做（我想要的是不同的信息：）。

更新：导致示例 URLIllegal request-target是：

http://host.host/hello\world
http://host.host/hello{all
http://host.host/hello"all

等等

我想使用 akka 流来将一些 json webservices 管道连接在一起。我想知道从 http 请求和流块到另一个流的最佳方法。有没有办法定义这样的图形并运行它而不是下面的代码？到目前为止，我尝试这样做，不确定它是否真的在流式传输：

override def receive: Receive = {
   case GetTestData(p, id) =>
     // Get the data and pipes it to itself through a message as recommended
     // https://doc.akka.io/docs/akka-http/current/client-side/request-level.html
     http.singleRequest(HttpRequest(uri = uri.format(p, id)))
       .pipeTo(self)

   case HttpResponse(StatusCodes.OK, _, entity, _) =>
     val initialRes = entity.dataBytes.via(JsonFraming.objectScanner(Int.MaxValue)).map(bStr => ChunkStreamPart(bStr.utf8String))

     // Forward the response to next job and pipes the request response to dedicated actor
     http.singleRequest(HttpRequest(
       method = HttpMethods.POST,
       uri = "googl.cm/flow",
       entity = HttpEntity.Chunked(ContentTypes.`application/json`, 
       initialRes)
     ))


   case resp …

升级到 Play 2.6 后，我在日志中看到了很多“EntityStreamException: Entity stream truncation”。它看起来与客户端超时密切相关，这并不令人担忧。对于这种常见的 HTTP 事件，在应用程序代码中出现异常冒泡有点令人担忧，所以我担心这不仅仅是客户端超时。

akka.http.scaladsl.model.EntityStreamException: Entity stream truncation
    at akka.http.impl.engine.parsing.HttpMessageParser$$anonfun$1.applyOrElse(HttpMessageParser.scala:321)
    at akka.http.impl.engine.parsing.HttpMessageParser$$anonfun$1.applyOrElse(HttpMessageParser.scala:319)
    at akka.stream.impl.fusing.Collect$$anon$6.$anonfun$wrappedPf$1(Ops.scala:217)
    at akka.stream.impl.fusing.SupervisedGraphStageLogic.withSupervision(Ops.scala:178)
    at akka.stream.impl.fusing.Collect$$anon$6.onPush(Ops.scala:219)
    at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:515)
    at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:478)
    at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:374)
    at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:588)
    at akka.stream.impl.fusing.GraphInterpreterShell$AsyncInput.execute(ActorGraphInterpreter.scala:474)
    at akka.stream.impl.fusing.GraphInterpreterShell.processEvent(ActorGraphInterpreter.scala:563)
    at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:730)
    at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:745)
    at akka.actor.Actor.aroundReceive(Actor.scala:517)
    at akka.actor.Actor.aroundReceive$(Actor.scala:515)
    at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:655)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
    at akka.actor.ActorCell.invoke(ActorCell.scala:561)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
    at akka.dispatch.Mailbox.run(Mailbox.scala:225)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
    at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

Is there a common approach to handle PATCH requests in REST API using circe library? By default, circe does not allow decoding partial JSON with only a part of the fields specified, i.e. it requires all fields to be set. You could use a withDefaults config, but it will be impossible to know if the field you received is null or just not specified. Here is a simplified sample of the possible solution. It uses Left[Unit] as a value to …

我刚开始学习 Scala、Akka Streams 和 Akka HTTP，所以如果问题太基本，请提前道歉。

我想在 HTTP 请求中执行一个 HTTP 请求，就像在下面的一段代码中一样：

  implicit val system = ActorSystem("ActorSystem")
  implicit val materializer = ActorMaterializer
  import system.dispatcher

  val requestHandler: Flow[HttpRequest, HttpResponse, _] = Flow[HttpRequest].map {
    case HttpRequest(HttpMethods.GET, Uri.Path("/api"), _, _, _) =>
      val responseFuture = Http().singleRequest(HttpRequest(uri = "http://www.google.com"))
      responseFuture.onComplete {
        case Success(response) =>
          response.discardEntityBytes()
          println(s"The request was successful")
        case Failure(ex) =>
          println(s"The request failed with: $ex")
      }
      //Await.result(responseFuture, 10 seconds)
      println("Reached HttpResponse")
      HttpResponse(
        StatusCodes.OK
      )
  }

  Http().bindAndHandle(requestHandler, "localhost", 8080)  

但在上述情况下，结果如下所示，这意味着Reached HttpResponse在完成请求之前首先到达：

Reached …