SCALA-355 Add code for Akka Stream article

Author: Matteo Di Pirro

build.sbt

@@ -217,7 +217,8 @@ lazy val scala_akka_2 = (project in file("scala-akka-2"))
       "com.lightbend.akka" %% "akka-stream-alpakka-sse" % "5.0.0",
       "com.typesafe.akka" %% "akka-persistence-typed" % AkkaVersion,
       "com.typesafe.akka" %% "akka-actor-testkit-typed" % AkkaVersion % Test,
-      "com.typesafe.akka" %% "akka-http-testkit" % AkkaHttpVersion % Test
+      "com.typesafe.akka" %% "akka-http-testkit" % AkkaHttpVersion % Test,
+      "com.typesafe.akka" %% "akka-stream-testkit" % AkkaVersion % Test
     ) ++ scalaTestDeps
   )
 val monocleVersion = "2.1.0"

scala-akka-2/src/main/scala/com/baeldung/scala/akka/stream/Main.scala

@@ -0,0 +1,20 @@
+package com.baeldung.scala.akka.stream
+
+import akka.actor.ActorSystem
+
+import scala.util.{Failure, Success}
+
+object Main extends App {
+  implicit val system: ActorSystem = ActorSystem("baeldung")
+
+  source
+    .via(parse)
+    .via(compare)
+    .runWith(sink)
+    .andThen {
+      case Failure(exception) => println(exception)
+      case Success((correct, total)) =>
+        println(s"$correct/$total correct answers")
+    }(system.dispatcher)
+    .onComplete(_ => system.terminate())(system.dispatcher)
+}

scala-akka-2/src/main/scala/com/baeldung/scala/akka/stream/package.scala

@@ -0,0 +1,29 @@
+package com.baeldung.scala.akka
+
+import akka.NotUsed
+import akka.stream.scaladsl.{Flow, Sink, Source}
+
+import scala.concurrent.Future
+
+package object stream {
+  val source: Source[String, NotUsed] = Source(
+    Seq("5,10", "15,15", "78,79", "12,12", "0,0", "456,456")
+  )
+
+  val parse: Flow[String, (Int, Int), NotUsed] =
+    Flow[String]
+      .map { pair =>
+        val parts = pair.split(",")
+        (parts(0).toInt, parts(1).toInt)
+      }
+
+  val compare: Flow[(Int, Int), Boolean, NotUsed] =
+    Flow[(Int, Int)]
+      .map { case (userAnswer, correctAnswer) => userAnswer == correctAnswer }
+
+  val sink: Sink[Boolean, Future[(Int, Int)]] = Sink.fold((0, 0)) {
+    case ((correctCount, total), wasCorrect) =>
+      if (wasCorrect) (correctCount + 1, total + 1)
+      else (correctCount, total + 1)
+  }
+}

scala-akka-2/src/test/scala/com/baeldung/scala/akka/stream/StreamTest.scala

@@ -0,0 +1,53 @@
+package com.baeldung.scala.akka.stream
+
+import akka.actor.ActorSystem
+import akka.stream.scaladsl.Keep
+import akka.stream.testkit.scaladsl.{TestSink, TestSource}
+import org.scalatest.concurrent.ScalaFutures.convertScalaFuture
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class StreamTest extends AnyFlatSpec with Matchers {
+  implicit val system: ActorSystem = ActorSystem("baeldung")
+
+  "The parse flow" should "parse pairs of integers" in {
+    val (pub, sub) = TestSource[String]()
+      .via(parse)
+      .toMat(TestSink[(Int, Int)]())(Keep.both)
+      .run()
+
+    pub.sendNext("1,1")
+    pub.sendNext("145,146")
+    pub.sendComplete()
+
+    sub.requestNext((1, 1))
+    sub.requestNext((145, 146))
+    sub.expectComplete()
+  }
+
+  "The compare flow" should "compare pairs of integers" in {
+    val (pub, sub) = TestSource[(Int, Int)]()
+      .via(compare)
+      .toMat(TestSink[Boolean]())(Keep.both)
+      .run()
+
+    pub.sendNext((1, 1))
+    pub.sendNext((145, 146))
+    pub.sendComplete()
+
+    sub.requestNext(true)
+    sub.requestNext(false)
+    sub.expectComplete()
+  }
+
+  "The sink sink" should "count the number of trues" in {
+    val (probe, result) = TestSource[Boolean]().toMat(sink)(Keep.both).run()
+
+    probe.sendNext(true)
+    probe.sendNext(false)
+    probe.sendNext(false)
+    probe.sendComplete()
+
+    result.futureValue shouldBe (1, 3)
+  }
+}