title: How do you throttle Flow in the latest Akka (2.4.6)?
body: <p>How do you throttle Flow in the latest Akka (2.4.6) ? I'd like to throttle Http client flow to limit number of requests to 3 requests per second. I found following example online but it's for old Akka and akka-streams API changed so much that I can't figure out how to  rewrite it.</p>
  
  <pre><code>def throttled[T](rate: FiniteDuration): Flow[T, T] = {
    val tickSource: Source[Unit] = TickSource(rate, rate, () =&gt; ())
    val zip = Zip[T, Unit]
    val in = UndefinedSource[T]
    val out = UndefinedSink[T]
    PartialFlowGraph { implicit builder =&gt;
      import FlowGraphImplicits._
      in ~&gt; zip.left ~&gt; Flow[(T, Unit)].map { case (t, _) =&gt; t } ~&gt; out
      tickSource ~&gt; zip.right
    }.toFlow(in, out)
  }
  </code></pre>
  
  <p>Here is my best attempt so far</p>
  
  <pre><code>def throttleFlow[T](rate: FiniteDuration) = Flow.fromGraph(GraphDSL.create() { implicit builder =&gt;
    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, _) =&gt; value }
    val messageExtractor = builder.add(map)
  
    val in = Inlet[T]("Req.in")
    val out = Outlet[T]("Req.out")
  
    out ~&gt; zip.in0
    ticker ~&gt; zip.in1
    zip.out ~&gt; messageExtractor.in
  
    FlowShape.of(in, messageExtractor.out)
  })
  </code></pre>
  
  <p>it throws exception in my main flow though :)</p>
  
  <pre><code>private val queueHttp = Source.queue[(HttpRequest, (Any, Promise[(Try[HttpResponse], Any)]))](1000, OverflowStrategy.backpressure)
    .via(throttleFlow(rate))
    .via(poolClientFlow)
    .mapAsync(4) {
      case (util.Success(resp), any) =&gt;
        val strictFut = resp.entity.toStrict(5 seconds)
        strictFut.map(ent =&gt; (util.Success(resp.copy(entity = ent)), any))
      case other =&gt;
        Future.successful(other)
    }
    .toMat(Sink.foreach({
      case (triedResp, (value: Any, p: Promise[(Try[HttpResponse], Any)])) =&gt;
        p.success(triedResp -&gt; value)
      case _ =&gt;
        throw new RuntimeException()
    }))(Keep.left)
    .run
  </code></pre>
  
  <p>where <code>poolClientFlow</code> is <code>Http()(system).cachedHostConnectionPool[Any](baseDomain)</code></p>
  
  <p>Exception is:</p>
  
  <pre><code>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)
  </code></pre>