Akka Stream - 简单的源/汇示例入口和出口不对应

Question

我开始学习Akka Stream。我有简化为这是一个问题：

import akka.actor.ActorSystem 
import akka.stream.{ActorMaterializer, ClosedShape} 
import akka.stream.scaladsl.{GraphDSL, RunnableGraph, Sink, Source} 

object Test extends App { 
    val graph = GraphDSL.create() { implicit b => 
    val in = Source.fromIterator(() => (1 to 10).iterator.map(_.toDouble)) 
    b.add(in) 
    val out = Sink.foreach[Double] { d => 
     println(s"elem: $d") 
    } 
    b.add(out) 
    in.to(out) 
    ClosedShape 
    } 

    implicit val system = ActorSystem() 
    implicit val mat = ActorMaterializer() 
    val rg = RunnableGraph.fromGraph(graph) 
    rg.run() 
} 

这将引发一个运行时异常：在线程“主” java.lang.IllegalArgumentException异常

例外：要求失败：入口[]和出口[]必须对应于入口[map.in]和插座[StatefulMapConcat.out]

---

的问题是，在我的实际情况下，我不能使用~>运营商从GraphDSL.Implicits，因为没有共同的超类型Source和Flow（我的图是从另一个DSL创建的，而不是在一个地方）。所以我只能使用b.add和in.to(out)。

Answer 1

似乎人们必须使用一个从builder.add获得出口的一个特殊的“复制”：

val graph = GraphDSL.create() { implicit b => 
    val in = Source.fromIterator(() => (1 to 10).iterator.map(_.toDouble)) 
    val out = Sink.foreach[Double] { d => 
     println(s"elem: $d") 
    } 
    import GraphDSL.Implicits._ 
    val inOutlet = b.add(in).out 
    // ... pass inOutlet around until ... 
    inOutlet ~> out 
    ClosedShape 
    }