我的喷雾 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 …

调用singleRequest时,如何自定义连接池使用的执行上下文？我简要地看了一下代码,并且调用singleRequest导致一条消息被发送到PoolMasterActor,后者又向池接口actor发送消息.

1. 每个连接是阻塞还是非阻塞？
2. 哪个上下文用于连接池？(我想确保我的HTTP请求不会阻止所有线程)

如何从 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

等等

我正在尝试使用Play和akka流创建一个简单的Websocket连接代理.交通流量是这样的:

(Client) request  ->         -> request (Server)
                      Proxy 
(Client) response <-         <- response (Server)

我按照一些例子后提出了以下代码:

def socket = WebSocket.accept[String, String] { request =>

val uuid = UUID.randomUUID().toString

// wsOut - actor that deals with incoming websocket frame from the Client
// wsIn - publisher of the frame for the Server
val (wsOut: ActorRef, wsIn: Publisher[String]) = {
  val source: Source[String, ActorRef] = Source.actorRef[String](10, OverflowStrategy.dropTail)
  val sink: Sink[String, Publisher[String]] = Sink.asPublisher(fanout = false)
  source.toMat(sink)(Keep.both).run()
}

// sink that deals with the …

嗨，有一个使用 Akka-Http 开发的服务。我必须向它添加 OAuth，根据文档，我正在使用authenticateOAuth2它。

但是代码没有编译并给出错误

类型不匹配，预期：(L) => server.Route，实际：(OauthInfo) => server.Route

我无法找到解决此问题的适当解决方案。我什至尝试了文档中示例中提到的确切代码，但它仍然抛出类似的编译时错误。

我正在使用带有圆圈的 akka-http。

这是我的代码：

def route(implicit system: ActorSystem, mat: ActorMaterializer): Route =
Route.seal {
  pathPrefix("newsletter-preferences") {
    authenticateOAuth2(realm = "Secure site", authenticator) { authInfo =>
      path("frequency" / LongNumber) { custNum =>
        authorize(hasScopes(authInfo)) {
          frequencyPreference(custNum) ~ addFreqPref(custNum)
        }
      } ~ path("pause" / LongNumber) { custNum =>
        authorize(hasScopes(authInfo)) {
          pauseInfo(custNum) ~ addPauseInfo(custNum) ~ unPauseUser(custNum)
        }
      }
    } ~
      path("health") {
        healthRoute()
      }
  }
}

def hasScopes(authInfo: OAuthInfo): …

我想使用 akka 流收听 websocket。也就是说，我想把它当作只是一个Source.

但是，所有官方示例都将 websocket 连接视为Flow.

我目前的方法是websocketClientFlow结合使用Source.maybe. TcpIdleTimeoutException当没有新的Messages 被发送到流中时，这最终会导致由于 a 的上游失败。

因此，我的问题是双重的：

1. 有没有办法——我显然错过了——把 websocket 当作一个Source?
2. 如果使用Flow是唯一的选择，如何TcpIdleTimeoutException正确处理？无法通过提供流监督策略来处理异常。使用 a 重新启动源RestartSource也无济于事，因为源不是问题所在。

更新

所以我尝试了两种不同的方法，为方便起见，将空闲超时设置为 1 秒

应用程序配置文件

akka.http.client.idle-timeout = 1s

使用 keepAlive（按照 Stefano 的建议）

Source.<Message>maybe()
    .keepAlive(Duration.apply(1, "second"), () -> (Message) TextMessage.create("keepalive"))
    .viaMat(Http.get(system).webSocketClientFlow(WebSocketRequest.create(websocketUri)), Keep.right())
    { ... }

执行此操作时，上游仍会失败并显示TcpIdleTimeoutException.

使用 RestartFlow

但是，我使用以下方法发现了这种方法RestartFlow：

final Flow<Message, Message, NotUsed> restartWebsocketFlow …

我想使用 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 …