Akka stream example 1

akka-streams-example-1.scala

import akka.Done
import akka.actor._
import akka.stream._
import akka.stream.scaladsl.Source

import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}


object TestStreamApp extends App {
  private implicit val executionContext = ExecutionContext.Implicits.global
  implicit val system = ActorSystem("TestSystem")
  implicit val materializer = ActorMaterializer()

  def process(page: Int): Future[Int] =
    Future.successful(page)
  
  val l: Stream[Int] = 1 #:: l.map(_ + 1)

  val streamedTasks: Future[Done] =
    Source(l) //toy example, streaming a list of int
      .takeWhile(_ <= 2)
      .mapAsync(parallelism = 2)(process)
      .map(s => s"Processed: $s")
      .runForeach(println)

  streamedTasks.andThen {
    case Success(_) => println("Completed tasks")
    case Failure(ex) => println("Failed to completed tasks", ex)
  }

}