How do I get instances of connected Inlet and Outl

Question

How do I get instances of connected Inlet and Outlet in FlowShape ? Consider following example

def throttleFlow[T](rate: FiniteDuration) = Flow.fromGraph(GraphDSL.create() { implicit builder =>
  import GraphDSL.Implicits._

  val ticker = Source.tick(rate, rate, Unit)

  val zip = builder.add(Zip[T, Unit.type])
  val map = Flow[(T, Unit.type)].map { case (value, _) => value }
  val messageExtractor = builder.add(map)

  val in = Inlet[T]("Req.in")
  val out = Outlet[T]("Req.out")

  out ~> zip.in0
  ticker ~> zip.in1
  zip.out ~> messageExtractor.in

  FlowShape.of(in, messageExtractor.out)
})

when I use it in Source.via() I get following exception

Caused by: java.lang.IllegalArgumentException: requirement failed: The output port [Req.out] is not part of the underlying graph.
    at scala.Predef$.require(Predef.scala:219)
    at akka.stream.impl.StreamLayout$Module$class.wire(StreamLayout.scala:204)

What am I missing ?

Answer 1

The in Inlet and out Outlet are not connected to anything. That is why there is an exception (unfortunately this kind of problem can only be detected at runtime)

You want a flow where the only open inlet is one of the zip inlets (zip.in0, since zip.in1 is connected to the ticker), and the only open outlet is the output of the messageExtractor, so how about this:

def throttleFlow[T](rate: FiniteDuration) = Flow.fromGraph(GraphDSL.create() { implicit builder =>
  import GraphDSL.Implicits._

  val ticker = Source.tick(rate, rate, ())

  val zip = builder.add(Zip[T, Unit])
  val map = Flow[(T, Unit)].map { case (value, _) => value }
  val messageExtractor = builder.add(map)

  ticker ~> zip.in1
  zip.out ~> messageExtractor.in

  FlowShape.of(zip.in0, messageExtractor.out)
})