akka-stream - 如何在流/图中以不同方式处理流的最后一个元素答案

【问题标题】：akka-stream - How to treat the last element of a stream differently in a Flow/Graphakka-stream - 如何在流/图中以不同方式处理流的最后一个元素
【发布时间】：2016-05-09 21:14:53
【问题描述】：

我正在尝试实现 Akka Streams Flow，它将 JSON 对象流转换为单个 JSON 对象数组的流。我可以使用Concat 在之前添加“[”和在之后添加“]”，以及使用Zip 在元素之间插入逗号，但我不知道如何不插入最后的逗号。

我目前的代码是：

trait JsonStreamSupport {

  protected def toJsonArrayString[T : Writes] = Flow[T].map(Json.toJson(_)).map(_.toString()).via(jsonArrayWrapper)


  private[this] val jsonArrayWrapper: Flow[String, String, NotUsed] = Flow.fromGraph(GraphDSL.create() { implicit b =>
    import GraphDSL.Implicits._
    val start = Source.single("[")
    val comma = Source.repeat(",")
    val end = Source.single("]")
    val concat = b.add(Concat[String](3))
    val zip = b.add(Zip[String,String])

    comma ~> zip.in1
    start ~> concat.in(0)
    zip.out.map({case (msg,delim) => msg + delim}) ~> concat.in(1)
    end ~> concat.in(2)
    FlowShape(zip.in0, concat.out)
  })
}

当前输出为：
[{"key":"value},{"key","value"},]
但我需要它是
[{"key":"value},{"key","value"}]（没有最后的逗号），
其中数组的每个元素仍然是流的不同元素，因此可以例如通过分块 HTTP 单独发送。

【问题讨论】：

标签： json scala akka akka-stream

【解决方案1】：

刚刚发现 intersperse 这正是您所需要的，而且比我最初建议的要简单得多：

http://doc.akka.io/api/akka/2.4.4/index.html#akka.stream.scaladsl.Flow@intersperse[T%3E:Out]%28start:T,inject:T,end:T%29:FlowOps.this.Repr[T]

【讨论】：