package sample.stream

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._

import scala.concurrent.Future

object SimpleStreamExample {

  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("Sys")
    import system.dispatcher

    implicit val materializer = ActorMaterializer()

    val numbers = 1 to 1000

    //We create a Source that will iterate over the number sequence
    val numbersSource: Source[Int, Unit] = Source.fromIterator(() => numbers.iterator)

    //Only let pass even numbers through the Flow
    val isEven: Flow[Int, Int, Unit] = Flow[Int].filter((num) => num % 2 == 0)

    //Create a Source of even random numbers by combining the random number Source with the even number filter Flow
    val evenNumbers: Source[Int, Unit] = numbersSource.via(isEven)

    //A Sink that will print out the data to the console output
    val consoleSink: Sink[Int, Future[Unit]] = Sink.foreach[Int](println)

    //Attach the evenNumbers Source with the console Sink. When the Sink is done shutdown the Actor System
    evenNumbers.runWith(consoleSink).onComplete(_ => system.shutdown())
  }
}