aludwiko

Source(List(1, 2, 3, 4))
 .map(_ * 2)
 .withAttributes(supervisionStrategy(Supervision.resumingDecider))
 .map(i => if (i == 2) throw new IllegalStateException("error"))
 .runWith(Sink.ignore)

aludwiko

Source(List(1, 2, 3, 4))
 .map(_ * 2)
 .withAttributes(supervisionStrategy(Supervision.resumingDecider))
 .map(i => if (i == 2) throw new IllegalStateException("error"))
 .runWith(Sink.ignore)

aludwiko

Source(infiniteTransactionsStream)
 .log(“got transaction”)

aludwiko

def saveHashAndReturnIt(hash: String): Future[String] = Future {
 if (Random.nextInt(5) == 1) {
   Thread.sleep(100000) //simulate database unavailability
 }
 hash
}

aludwiko

Source(1 to 5)
 .alsoTo(Sink.foreach{x =>
   Thread.sleep(100)
   println(s”elem:$x alsoTo”)}
 )
 .map { x =>
   println(s”elem:$x map”)
   x
 }
 .to(Sink.foreach(x => println(s”elem:$x to”)))

aludwiko

val roomsTemperatureReadings =
 Stream.continually(TemperatureReading(randomRoom, randomTemperature))

Source(roomsTemperatureReadings)
 .alsoTo(Sink.foreach(reading => createRoomIfNotExist(reading.roomId)))
 .to(Sink.foreach(saveReading))
 .run()

aludwiko

Source(List("1","2","a","4","5"))
 .map(_.toInt)
 .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
 .runForeach(println)

aludwiko

val nonLinearFlow = Flow.fromGraph(GraphDSL.create() { implicit builder =>
  
 import GraphDSL.Implicits._
  
 val dangerousFlow = Flow[Int].map {
   case 5 => throw new RuntimeException(“BOOM!”)
   case x => x
 }
  
 val safeFlow = Flow[Int]

aludwiko

Source(1 to 10)
 .via(nonLinearFlow)
 .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
 .runForeach(println)

aludwiko

val dangerousFlow = Flow[Int].map {
   case 5 => Failure(new RuntimeException(“BOOM!”))
   case x => Try(x)
}

val safeFlow = Flow[Int].map(Try(_))
val bcast = builder.add(Broadcast[Int](2))
val zip = builder.add(Zip[Try[Int], Try[Int]])