CODEWITHSUNDEEP
scalaakka-stream
0__
0__

Reputation: 67280

Akka Stream - simple source/sink example inlets and outlets do not correspond

I'm starting to learn Akka Stream. I have a problem that I simplified to this:

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()
}

This throws a runtime exception:

Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: The inlets [] and outlets [] must correspond to the inlets [map.in] and outlets [StatefulMapConcat.out]

The problem is, in my actual case I cannot use the ~> operator from GraphDSL.Implicits, because there is no common super-type of Source and Flow (my graph is created from another DSL and not in one single place). So I can only use b.add and in.to(out).

Upvotes: 2

Views: 952

Answers (1)

0__
0__

Reputation: 67280

It seems one has to use a special "copy" of the outlet that one gets from 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
  }

Upvotes: 2

Related Questions