How do you throttle Flow in the latest Akka (2.4.6)?

Question

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.

def throttled[T](rate: FiniteDuration): Flow[T, T] = {
  val tickSource: Source[Unit] = TickSource(rate, rate, () => ())
  val zip = Zip[T, Unit]
  val in = UndefinedSource[T]
  val out = UndefinedSink[T]
  PartialFlowGraph { implicit builder =>
    import FlowGraphImplicits._
    in ~> zip.left ~> Flow[(T, Unit)].map { case (t, _) => t } ~> out
    tickSource ~> zip.right
  }.toFlow(in, out)
}

Here is my best attempt so far

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

it throws exception in my main flow though :)

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) =>
      val strictFut = resp.entity.toStrict(5 seconds)
      strictFut.map(ent => (util.Success(resp.copy(entity = ent)), any))
    case other =>
      Future.successful(other)
  }
  .toMat(Sink.foreach({
    case (triedResp, (value: Any, p: Promise[(Try[HttpResponse], Any)])) =>
      p.success(triedResp -> value)
    case _ =>
      throw new RuntimeException()
  }))(Keep.left)
  .run

where poolClientFlow is Http()(system).cachedHostConnectionPool[Any](baseDomain)

Exception is:

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)

The above is pretty easy to redo in new akka streams. But it also seems flawed to me. Zip requires one from each source. A better approach might be to merge the requests and the ticks, and then have a stateful BidiFlow that answers requests with a 503 if there are more requests than ticks. — Rüdiger Klaehn
@RüdigerKlaehn Could you please show your version ? I've updated question with my non-working version. I don't want requests to fail. I want them to be queued and wait using backpressure. — expert
@Qingwei , the problem is that expert apparently wants to throttle globally (over multiple materializations of a flow), so throttle won't work. — Rüdiger Klaehn

Rüdiger Klaehn Rüdiger Klaehn · Accepted Answer · 2016-05-24T22:02:29

Here is an attempt that uses the throttle method as mentioned by @Qingwei. The key is to not use bindAndHandle(), but to use bind() and throttle the flow of incoming connections before handling them. The code is taken from the implementation of bindAndHandle(), but leaves out some error handling for simplicity. Please don't do that in production.

implicit val system = ActorSystem("test")
implicit val mat = ActorMaterializer()
import system.dispatcher
val maxConcurrentConnections = 4

val handler: Flow[HttpRequest, HttpResponse, NotUsed] = complete(LocalDateTime.now().toString)

def handleOneConnection(incomingConnection: IncomingConnection): Future[Done] =
  incomingConnection.flow
        .watchTermination()(Keep.right)
        .joinMat(handler)(Keep.left)
        .run()

Http().bind("127.0.0.1", 8080)
  .throttle(3, 1.second, 1, ThrottleMode.Shaping)
  .mapAsyncUnordered(maxConcurrentConnections)(handleOneConnection)
  .to(Sink.ignore)
  .run()

How do you throttle Flow in the latest Akka (2.4.6)?

1 Answers