raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

x-ref to queue.md

Browse source
This commit is contained in:
Ignasi Marimon-Clos 2020-06-08 20:13:29 +02:00
parent a72ee4d5b8
commit 506cbc5bf4
3 changed files with 35 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

3
akka-docs/src/main/paradox/stream/operators/Source-or-Flow/throttle.md
View file

@ -14,6 +14,9 @@ Limit the throughput to a specific number of elements per time unit, or a specif
Limit the throughput to a specific number of elements per time unit, or a specific total cost per time unit, where
a function has to be provided to calculate the individual cost of each element.
The throttle operator combines will with the @ref[`queue`](./../Source/queue.md) operator to adapt the speeds on both ends of the `queue`-`throttle` pair.
See also @ref:[Buffers and working with rate](../../stream-rate.md) for related operators.
## Example

49
akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Throttle.java
View file

@ -1,3 +1,7 @@
/*
* Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
*/
package jdocs.stream.operators.sourceorflow;
import akka.NotUsed;
@ -11,39 +15,26 @@ import java.time.Duration;
import java.util.stream.Stream;
import docs.stream.operators.sourceorflow.ThrottleCommon.Frame;
/**
*
*/
/** */
public class Throttle {
public static void main(String[] args) {
ActorSystem actorSystem = ActorSystem.create("25fps-throttled-stream");
Materializer mat = Materializer.matFromSystem(actorSystem);
public static void main(String[] args) {
ActorSystem actorSystem = ActorSystem.create("25fps-throttled-stream");
Materializer mat = Materializer.matFromSystem(actorSystem);
Source<Frame, NotUsed> frameSource =
Source
.fromIterator(() -> Stream.iterate(0 , i -> i+1).iterator())
.map(i -> new Frame(i.intValue())) ;
Source<Frame, NotUsed> frameSource =
Source.fromIterator(() -> Stream.iterate(0, i -> i + 1).iterator())
.map(i -> new Frame(i.intValue()));
// #throttle
int framesPerSecond = 24;
Source<Frame, NotUsed> videoThrottling = frameSource
.throttle(
framesPerSecond,
Duration.ofSeconds(1),
framesPerSecond * 30,
ThrottleMode.shaping()
);
// serialize `Frame` and send over the network.
// #throttle
videoThrottling
.map(f -> f.i())
.to(Sink.foreach(System.out::println))
.run(mat);
}
// #throttle
int framesPerSecond = 24;
Source<Frame, NotUsed> videoThrottling =
frameSource.throttle(
framesPerSecond, Duration.ofSeconds(1), framesPerSecond * 30, ThrottleMode.shaping());
// serialize `Frame` and send over the network.
// #throttle
videoThrottling.map(f -> f.i()).to(Sink.foreach(System.out::println)).run(mat);
}
}

22
akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Throttle.scala
View file

@ -1,3 +1,7 @@
/*
* Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
*/
package docs.stream.operators.sourceorflow
import akka.NotUsed
@ -22,22 +26,18 @@ object Throttle extends App {
val framesPerSecond = 24
// val frameSource: Source[Frame,_]
val videoThrottling = frameSource
.throttle(
framesPerSecond,
1.second,
framesPerSecond * 30, // maximumBurst
ThrottleMode.shaping
)
val videoThrottling = frameSource.throttle(
framesPerSecond,
1.second,
framesPerSecond * 30, // maximumBurst
ThrottleMode.shaping)
// serialize `Frame` and send over the network.
// #throttle
videoThrottling
.to(Sink.foreach(println))
.run()
videoThrottling.to(Sink.foreach(println)).run()
}
object ThrottleCommon{
object ThrottleCommon {
// used in ThrottleJava
case class Frame(i: Int)