1 - triageddiscusshelp wantedt:http:marshallingt:routingt:stream
仓库指标
- 星标
- (1,311 个星标)
- PR 合并指标
- (平均合并 1天 10小时) (30 天内合并 2 个 PR)
描述
Streams are great but I got a stream of thousand elements. Unfortunately every element is emitted in one http chunk (and the elements are small). Would be nice if the framework supports "merging" of multiple elements into one chunk and return a valid json array of elements.
My workaround
def toJsonBytes(source: Seq[T]): ByteString =
ByteString(source.map(t => t.asJson.noSpaces).mkString(","))
// emit 100 elements in one http chunk
val chunk = t.groupedWithin(100, 100.millis)
val fst = Flow[Seq[T]].map(toJsonBytes)
val rst = Flow[Seq[T]].map(ByteString(",") ++ toJsonBytes(_))
val together = chunk.prefixAndTail(1).flatMapConcat {
case (head, tail) =>
Source(head).via(fst).concat(tail.via(rst))
}
complete(
HttpEntity(
ContentTypes.`application/json`,
Source.single(ByteString("[")) ++ together ++ Source.single(ByteString("]"))
)
)