使用 akka http 分块响应流式传输 JSON 对象答案

【问题标题】：Streaming JSON objects using akka http chunked response使用 akka http 分块响应流式传输 JSON 对象
【发布时间】：2023-04-01 23:37:01
【问题描述】：

像这样使用 akka-http 是滥用还是有某种危险？

在服务器上

def source(consumerOffset: UUID) = 
  readJournal.eventsByTag(“MyTag", consumerOffset).map(_.asJson)

pathPrefix("stream" / Segment.map(UUID.fromString)) { offset =>
  pathEndOrSingleSlash {
    get {
      complete {
        HttpResponse(
          StatusCodes.OK,
          entity = HttpEntity(ContentTypes.`application/json`, source(offset))
        )
      }
    }
  }
}

然后在客户端

Source.single(HttpRequest("http://localhost:9000/stream"))
  .mapAsync(1) { r =>
    Http().singleRequest(r).map { res =>
      res.entity.dataBytes.map(_.parse[Event])
    }
  }
  .flatMapConcat(identity).mapAsync(processEvent)

UPD：

是否保证我发送的块在客户端是相同的。
是否可以响应可能无穷无尽的块？
这种响应的正确 Content-Type 是什么？

UPD 2：

Akka 2.4.9 增加了响应流的能力。并且基本上完全一样，提供了一些语法糖。请参阅docs。

【问题讨论】：

你关心哪一部分？乍一看，这一切似乎都是锅炉板......
1.是否保证我发送的块在客户端是相同的。
2.是否可以对可能无穷无尽的块进行响应？
我已经更新了问题。
@DenisMikhaylov 除此之外，如果客户端以黑白方式关闭连接会发生什么，但服务器无法知道它，服务器认为客户端只是消耗缓慢并且它保持缓冲响应。如何处理这些场景？

标签： akka akka-stream akka-http

【解决方案1】：

按顺序回答您的问题：

是的，客户端将收到BytesString 以 json 编码的对象的表示形式。
是的，可以在服务器端有一个永不终止的流源。
您在问题 (application/json) 中指定的内容是正确的。

【讨论】：