有没有像 SinkSource[T] 这样的东西？答案

【问题标题】：Is there something like a SinkSource[T]?有没有像 SinkSource[T] 这样的东西？
【发布时间】：2015-01-27 16:01:31
【问题描述】：

我正在寻找提供Sink 和Source 的SinkSource。如果元素流入Sink，则应在相应的Source 中提供它。下面的代码说明了我的意思：

object SinkSource {
  def apply[T] = new {
    def sink: Sink[T] = ???
    def source: Source[T] = ???
  }
}
val flowgraph = FlowGraph { implicit fgb =>
  import FlowGraphImplicits._
  val sinksource = SinkSource[Int]
  Source(1 to 5) ~> sinksource.sink
                    sinksource.source ~> Sink.foreach(print)
}
implicit val actorSystem = ActorSystem(name = "System")
implicit val flowMaterializer = FlowMaterializer()
val materializedMap = flowgraph.run()

如果执行，应该打印：12345
那么，SinkSource 是否存在（在 API 中没有看到它）或者有谁知道如何实现它？我应该提一下，我需要对Sink 和Source 进行不同的访问，这样Flow 就不是这种特殊形式的解决方案：

Source(1 to 5) ~> Flow[Int] ~> Sink.foreach(println)

【问题讨论】：

标签： scala akka-stream

【解决方案1】：

通常，如果已经提出问题，就会想到一些想法：事实证明，我不需要 Sink 和 Source、JunctionInPort 和 JunctionOutPort 就足够了。
就这样吧：

object SinkSource {
  def apply[T](implicit fgb: FlowGraphBuilder) = new SinkSource[T]
}
class SinkSource[T](implicit fgb: FlowGraphBuilder) {
  import FlowGraphImplicits._
  private val merge = Merge[T]
  private val bcast = Broadcast[T]
  Source.empty ~> merge
  merge ~> bcast
  bcast ~> Sink.ignore
  def in: JunctionInPort[T] = merge
  def out: JunctionOutPort[T] = bcast
}
val flowgraph = FlowGraph { implicit fgb =>
  import FlowGraphImplicits._
  val source = Source(1 to 5)
  val sink = Sink.foreach(println)
  val sinkSource = SinkSource[Int]
  source ~> sinkSource.in
            sinkSource.out ~> sink
}
implicit val actorSystem = ActorSystem(name = "System")
implicit val flowMaterializer = FlowMaterializer()
val materializedMap = flowgraph.run()

【讨论】：