example for some sink operators akka#25468

Examples for takeLast, last and lastOption operators
dwijnand · Oct 25, 2018 · ba996f8 · ba996f8
1 parent 91b0c38
commit ba996f8
Show file tree

Hide file tree

Showing 6 changed files with 124 additions and 19 deletions.
diff --git a/akka-docs/src/main/paradox/stream/operators/Sink/last.md b/akka-docs/src/main/paradox/stream/operators/Sink/last.md
@@ -26,4 +26,10 @@ completes. If the stream completes with no elements the @scala[`Future`] @java[`
 
 @@@
 
+## Example
 
+Scala
+:   @@snip [LastSinkSpec.scala](/akka-stream-tests/src/test/scala/akka/stream/scaladsl/LastSinkSpec.scala) { #last-operator-example }
+
+Java
+:   @@snip [SinkDocExamples.java](/akka-docs/src/test/java/jdocs/stream/operators/SinkDocExamples.java) { #last-operator-example }
diff --git a/akka-docs/src/main/paradox/stream/operators/Sink/lastOption.md b/akka-docs/src/main/paradox/stream/operators/Sink/lastOption.md
@@ -27,3 +27,10 @@ completed with @scala[`None`] @java[an empty `Optional`].
 
 @@@
 
+## Example
+
+Scala
+:   @@snip [LastSinkSpec.scala](/akka-stream-tests/src/test/scala/akka/stream/scaladsl/LastSinkSpec.scala) { #lastOption-operator-example }
+
+Java
+:   @@snip [SinkDocExamples.java](/akka-docs/src/test/java/jdocs/stream/operators/SinkDocExamples.java) { #lastOption-operator-example }
diff --git a/akka-docs/src/main/paradox/stream/operators/Sink/takeLast.md b/akka-docs/src/main/paradox/stream/operators/Sink/takeLast.md
@@ -28,4 +28,10 @@ If there is a failure signaled in the stream the @scala[`Future`] @java[`Complet
 
 @@@
 
+## Example
 
+Scala
+:   @@snip [TakeLastSinkSpec.scala](/akka-stream-tests/src/test/scala/akka/stream/scaladsl/TakeLastSinkSpec.scala) { #takeLast-operator-example }
+
+Java
+:   @@snip [SinkDocExamples.java](/akka-docs/src/test/java/jdocs/stream/operators/SinkDocExamples.java) { #takeLast-operator-example }
diff --git a/akka-docs/src/test/java/jdocs/stream/operators/SinkDocExamples.java b/akka-docs/src/test/java/jdocs/stream/operators/SinkDocExamples.java
@@ -6,30 +6,75 @@
 
 import akka.NotUsed;
 import akka.actor.ActorSystem;
+
 import akka.stream.ActorMaterializer;
 import akka.stream.Materializer;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
-
-import java.util.Arrays;
+//#takeLast-operator-example
+import akka.japi.Pair;
+//#takeLast-operator-example
+import java.util.*;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 public class SinkDocExamples {
 
+    private final static ActorSystem system = ActorSystem.create("SourceFromExample");
+    private final static Materializer materializer = ActorMaterializer.create(system);
+
     static void reduceExample() throws InterruptedException, ExecutionException, TimeoutException {
 
-        final ActorSystem system = ActorSystem.create("SourceFromExample");
-        final Materializer materializer = ActorMaterializer.create(system);
         //#reduce-operator-example
         Source<Integer, NotUsed> ints = Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
         CompletionStage<Integer> sum = ints.runWith(Sink.reduce((a, b) -> a + b), materializer);
-        int result = sum.toCompletableFuture().get(3, TimeUnit.SECONDS);
-        System.out.println(result);
+        sum.thenAccept(System.out::println);
         // 55
         //#reduce-operator-example
     }
-}
 
+    static void takeLastExample() throws InterruptedException, ExecutionException, TimeoutException {
+        //#takeLast-operator-example
+        // pair of (Name, GPA)
+        List<Pair> sortedStudents = Arrays.asList(new Pair<>("Benita", 2.1), new Pair<>("Adrian", 3.1),
+                new Pair<>("Alexis", 4), new Pair<>("Kendra", 4.2), new Pair<>("Jerrie", 4.3), new Pair<>("Alison", 4.7));
+
+        Source<Pair, NotUsed> studentSource = Source.from(sortedStudents);
+
+        CompletionStage<List<Pair>> topThree = studentSource.runWith(Sink.takeLast(3), materializer);
+
+        topThree.thenAccept(result -> {
+            System.out.println("#### Top students ####");
+            for (int i = result.size() - 1; i >= 0; i--) {
+                Pair<String, Double> s = result.get(i);
+                System.out.println("Name: " + s.first() + ", " + "GPA: " + s.second());
+            }
+        });
+        /*
+        #### Top students ####
+        Name: Alison, GPA: 4.7
+        Name: Jerrie, GPA: 4.3
+        Name: Kendra, GPA: 4.2
+      */
+        //#takeLast-operator-example
+    }
+
+    static void lastExample() throws InterruptedException, ExecutionException, TimeoutException {
+        //#last-operator-example
+        Source<Integer, NotUsed> source = Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
+        CompletionStage<Integer> result = source.runWith(Sink.last(), materializer);
+        result.thenAccept(System.out::println);
+        // 10
+        //#last-operator-example
+    }
+
+    static void lastOptionExample() throws InterruptedException, ExecutionException, TimeoutException {
+        //#lastOption-operator-example
+        Source<Integer, NotUsed> source = Source.empty();
+        CompletionStage<Optional<Integer>> result = source.runWith(Sink.lastOption(), materializer);
+        result.thenAccept(System.out::println);
+        // Optional.empty
+        //#lastOption-operator-example
+    }
+}
diff --git a/akka-stream-tests/src/test/scala/akka/stream/scaladsl/LastSinkSpec.scala b/akka-stream-tests/src/test/scala/akka/stream/scaladsl/LastSinkSpec.scala
@@ -4,25 +4,30 @@
 
 package akka.stream.scaladsl
 
-import scala.concurrent.Await
-import scala.concurrent.duration._
-
-import akka.stream.ActorMaterializer
-import akka.stream.ActorMaterializerSettings
+import akka.stream.{ ActorMaterializer, ActorMaterializerSettings }
 import akka.stream.testkit._
-import akka.stream.testkit.Utils._
 import akka.stream.testkit.scaladsl.StreamTestKit._
 
+import scala.concurrent.duration._
+import scala.concurrent.{ Await, Future }
+
 class LastSinkSpec extends StreamSpec with ScriptedTest {
 
   val settings = ActorMaterializerSettings(system)
 
-  implicit val materializer = ActorMaterializer(settings)
+  implicit val materializer: ActorMaterializer = ActorMaterializer(settings)
+  implicit val ec = system.dispatcher
 
   "A Flow with Sink.last" must {
 
-    "yield the last value" in assertAllStagesStopped {
-      Await.result(Source(1 to 42).map(identity).runWith(Sink.last), 1.second) should be(42)
+    "yield the last value" in {
+      //#last-operator-example
+      val source = Source(1 to 10)
+      val result: Future[Int] = source.runWith(Sink.last)
+      result.map(println)
+      // 10
+      //#last-operator-example
+      result.futureValue shouldEqual 10
     }
 
     "yield the first error" in assertAllStagesStopped {
@@ -52,8 +57,14 @@ class LastSinkSpec extends StreamSpec with ScriptedTest {
       } should be theSameInstanceAs (ex)
     }
 
-    "yield None for empty stream" in assertAllStagesStopped {
-      Await.result(Source.empty[Int].runWith(Sink.lastOption), 1.second) should be(None)
+    "yield None for empty stream" in {
+      //#lastOption-operator-example
+      val source = Source.empty[Int]
+      val result: Future[Option[Int]] = source.runWith(Sink.lastOption)
+      result.map(println)
+      // None
+      //#lastOption-operator-example
+      result.futureValue shouldEqual None
     }
 
   }

diff --git a/akka-stream-tests/src/test/scala/akka/stream/scaladsl/TakeLastSinkSpec.scala b/akka-stream-tests/src/test/scala/akka/stream/scaladsl/TakeLastSinkSpec.scala
@@ -25,6 +25,36 @@ class TakeLastSinkSpec extends StreamSpec {
       result should be(Seq(4, 5, 6))
     }
 
+    "return top three student based on GPA correctly" in {
+      implicit val ex = system.dispatcher
+      //#takeLast-operator-example
+      case class Student(name: String, gpa: Double)
+
+      val students = List(Student("Alison", 4.7), Student("Adrian", 3.1), Student("Alexis", 4),
+        Student("Benita", 2.1), Student("Kendra", 4.2), Student("Jerrie", 4.3)).sortBy(_.gpa)
+
+      val sourceOfStudents = Source(students)
+
+      val result: Future[Seq[Student]] = sourceOfStudents.runWith(Sink.takeLast(3))
+
+      result.foreach { topThree ⇒
+        println("#### Top students ####")
+        topThree.reverse foreach { s ⇒
+          println(s"Name: ${s.name}, GPA: ${s.gpa}")
+        }
+      }
+      /*
+          #### Top students ####
+          Name: Alison, GPA: 4.7
+          Name: Jerrie, GPA: 4.3
+          Name: Kendra, GPA: 4.2
+      */
+
+      //#takeLast-operator-example
+
+      result.futureValue shouldEqual students.takeRight(3)
+    }
+
     "return the number of elements taken when the stream completes" in {
       val input = 1 to 4
       val future: Future[immutable.Seq[Int]] = Source(input).runWith(Sink.takeLast(5))