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

Headers and formats

Browse source
This commit is contained in:
Ignasi Marimon-Clos 2020-07-07 14:58:56 +02:00
parent 0e3b4605ad
commit 024710a6e0
2 changed files with 61 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

109
akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Monitor.java
View file

@ -1,3 +1,7 @@
/*
* Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
*/
package jdocs.stream.operators.sourceorflow;
import akka.Done;
@ -16,66 +20,61 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
*
*/
/** */
public class Monitor {
// #monitor
private static <T> void printMonitorState(FlowMonitorState.StreamState<T> state) {
if (FlowMonitorState.Finished$.class.isInstance(state)) {
System.out.println("Stream is initialized but hasn't processed any element");
} else if (FlowMonitorState.Received.class.isInstance(state)) {
FlowMonitorState.Received msg = (FlowMonitorState.Received) state;
System.out.println("Last message received: " + msg.msg());
} else if (FlowMonitorState.Failed.class.isInstance(state)) {
Throwable cause = ((FlowMonitorState.Failed) state).cause();
System.out.println("Stream failed with cause: " + cause.getMessage());
} else {
System.out.println("Stream completed already");
}
// #monitor
private static <T> void printMonitorState(FlowMonitorState.StreamState<T> state) {
if (FlowMonitorState.Finished$.class.isInstance(state)) {
System.out.println("Stream is initialized but hasn't processed any element");
} else if (FlowMonitorState.Received.class.isInstance(state)) {
FlowMonitorState.Received msg = (FlowMonitorState.Received) state;
System.out.println("Last message received: " + msg.msg());
} else if (FlowMonitorState.Failed.class.isInstance(state)) {
Throwable cause = ((FlowMonitorState.Failed) state).cause();
System.out.println("Stream failed with cause: " + cause.getMessage());
} else {
System.out.println("Stream completed already");
}
}
// #monitor
public static void main(String[] args)
throws InterruptedException, TimeoutException, ExecutionException {
ActorSystem actorSystem = ActorSystem.create("25fps-stream");
// #monitor
Source<Integer, FlowMonitor<Integer>> monitoredSource =
Source.fromIterator(() -> Arrays.asList(0, 1, 2, 3, 4, 5).iterator())
.throttle(5, Duration.ofSeconds(1))
.monitorMat(Keep.right());
Pair<FlowMonitor<Integer>, CompletionStage<Done>> run =
monitoredSource.toMat(Sink.foreach(System.out::println), Keep.both()).run(actorSystem);
FlowMonitor<Integer> monitor = run.first();
// if we peek on the stream too early it probably won't have processed any element.
printMonitorState(monitor.state());
// #monitor
// exclude from rendered snippet
Thread.sleep(500);
// #monitor
// ...
// sometime later, our code has progressed. We can peek in the stream
// again to see what's the latest element processed
printMonitorState(monitor.state());
public static void main(String[] args) throws InterruptedException, TimeoutException, ExecutionException {
ActorSystem actorSystem = ActorSystem.create("25fps-stream");
// #monitor
// exclude from rendered snippet
run.second().toCompletableFuture().get(1, TimeUnit.SECONDS);
// #monitor
// #monitor
// Eventually, the stream completes and if we check the state it reports the streasm finished.
printMonitorState(monitor.state());
// #monitor
// #monitor
Source<Integer, FlowMonitor<Integer>> monitoredSource =
Source
.fromIterator(() -> Arrays.asList(0, 1, 2, 3, 4, 5).iterator())
.throttle(5, Duration.ofSeconds(1))
.monitorMat(Keep.right());
Pair<FlowMonitor<Integer>, CompletionStage<Done>> run = monitoredSource
.toMat(Sink.foreach(System.out::println), Keep.both()).run(actorSystem);
FlowMonitor<Integer> monitor = run.first();
// if we peek on the stream too early it probably won't have processed any element.
printMonitorState(monitor.state());
// #monitor
// exclude from rendered snippet
Thread.sleep(500);
// #monitor
// ...
// sometime later, our code has progressed. We can peek in the stream
// again to see what's the latest element processed
printMonitorState(monitor.state());
// #monitor
// exclude from rendered snippet
run.second().toCompletableFuture().get(1, TimeUnit.SECONDS);
// #monitor
// #monitor
// Eventually, the stream completes and if we check the state it reports the streasm finished.
printMonitorState(monitor.state());
// #monitor
run.second().toCompletableFuture().whenComplete((x, t) -> actorSystem.terminate());
}
run.second().toCompletableFuture().whenComplete((x, t) -> actorSystem.terminate());
}
}

15
akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Monitor.scala
View file

@ -1,3 +1,7 @@
/*
* Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
*/
package docs.stream.operators.sourceorflow
import akka.Done
@ -15,7 +19,7 @@ import scala.concurrent.Future
import scala.concurrent.duration._
/**
*
*
*/
class Monitor {
@ -38,14 +42,9 @@ class Monitor {
}
}
val monitoredSource: Source[Int, FlowMonitor[Int]] = source
.take(6)
.throttle(5, 1.second)
.monitorMat(Keep.right)
val monitoredSource: Source[Int, FlowMonitor[Int]] = source.take(6).throttle(5, 1.second).monitorMat(Keep.right)
val monitoredStream: (FlowMonitor[Int], Future[Done]) =
monitoredSource
.toMat(Sink.foreach(println))(Keep.both)
.run()
monitoredSource.toMat(Sink.foreach(println))(Keep.both).run()
val flowMonitor = monitoredStream._1