I would add though, that it is generally not recommended to set the maximumBurst to zero as it is highly inefficient (needs to contact the Scheduler for basically all messages) and for higher rates it has not enough precision. So maximumBurst is also used to be able to compensate for the scheduler inaccuracy and increase performance (and hence being able to actually meet higher rates).
-Endre On Wed, May 25, 2016 at 8:16 AM, <[email protected]> wrote: > 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 >> 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]) 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. > > > -- > >>>>>>>>>> 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.
