Could you please explain meaning of `maximumBurst` parameter with example? It's unclear to me from ScalaDoc.
On Wednesday, May 25, 2016 at 1:17:45 AM UTC+3, Konrad Malawski wrote: > > 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 <http://akka.io> @ Lightbend <http://lightbend.com> > > On 25 May 2016 at 00:12:32, Jack Daniels ([email protected] > <javascript:>) 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] <javascript:>. > To post to this group, send email to [email protected] > <javascript:>. > 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.
