> On 25 May 2016, at 00:33, Jack Daniels <[email protected]> wrote: > > Could you please explain meaning of `maximumBurst` parameter with example?
If you throttle to a certain rate, `maximumBurst` defines what should happen if fewer elements have arrived from upstream for some time (your bucket contains some tokens) and now upstream presses hard again. If you set `maximumBurst` to 0 or 1, the rate will never be exceeded, at no time. If you set it to a higher level, the rate is more of an average, because you allow short bursts of limited (by the minimum of `maximumBurst` and available tokens) size. Example: You define a rate of 1 element per second. For 10 seconds no elements from upstream, hence 10 tokens in the bucket. Then “unlimited” elements from upstream. With a `maximumBurst` of 2, downstream will get 2 elements per second until all tokes have been used up (after 7.5 seconds). Then the rate will be 1 element per second. Makes sense? > 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 > <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 > <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/ <http://akka.io/docs/> >> >>>>>>>>>> Check the FAQ: >> >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >> >>>>>>>>>> <http://doc.akka.io/docs/akka/current/additional/faq.html> >> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user >> >>>>>>>>>> <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 >> <https://groups.google.com/group/akka-user>. >> For more options, visit https://groups.google.com/d/optout >> <https://groups.google.com/d/optout>. > > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ <http://akka.io/docs/> > >>>>>>>>>> Check the FAQ: > >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html > >>>>>>>>>> <http://doc.akka.io/docs/akka/current/additional/faq.html> > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > >>>>>>>>>> <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] > <mailto:[email protected]>. > To post to this group, send email to [email protected] > <mailto:[email protected]>. > Visit this group at https://groups.google.com/group/akka-user > <https://groups.google.com/group/akka-user>. > For more options, visit https://groups.google.com/d/optout > <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.
signature.asc
Description: Message signed with OpenPGP using GPGMail
