Throttle a Flow? Sorry, I did not dive deep into your code, as perhaps there's no need to, because:
http://doc.akka.io/docs/akka/2.4.6/scala/stream/stages-overview.html#throttle and throttle on http://doc.akka.io/api/akka/2.4.6/#akka.stream.scaladsl.FlowOps -- Konrad `ktoso` Malawski Akka @ Lightbend On 25 May 2016 at 00:12:32, Jack Daniels ([email protected]) wrote: Hi guys! 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 :) It looks like this 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) It's totally confusing to me. How do I get connected inlet and outlet ? I thought Akka is supposed to connect them when I use my flow in .via() call. Thanks! -- >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>>>>>>>> Check the FAQ: >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user --- You received this message because you are subscribed to the Google Groups "Akka User List" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. To post to this group, send email to [email protected]. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout. -- >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>>>>>>>> Check the FAQ: >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user --- You received this message because you are subscribed to the Google Groups "Akka User List" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. To post to this group, send email to [email protected]. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
