Andrzej Ludwikowski aludwiko

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

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

## akka_streams_pitfalls_7_2.scala
Source(infiniteTransactionsStream)
 .log(“got transaction”)

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

## akka_streams_pitfalls_8.scala
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”)))

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

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

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

## akka_streams_pitfalls_9_1.scala
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]

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

## akka_streams_pitfalls_9_3.scala
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]])
	Source(List(1, 2, 3, 4))
	.map(_ * 2)
	.withAttributes(supervisionStrategy(Supervision.resumingDecider))
	.map(i => if (i == 2) throw new IllegalStateException("error"))
	.runWith(Sink.ignore)
	Source(infiniteTransactionsStream)
	.log(“got transaction”)
	def saveHashAndReturnIt(hash: String): Future[String] = Future {
	if (Random.nextInt(5) == 1) {
	Thread.sleep(100000) //simulate database unavailability
	}
	hash
	}
	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”)))
	val roomsTemperatureReadings =
	Stream.continually(TemperatureReading(randomRoom, randomTemperature))

	Source(roomsTemperatureReadings)
	.alsoTo(Sink.foreach(reading => createRoomIfNotExist(reading.roomId)))
	.to(Sink.foreach(saveReading))
	.run()
	Source(List("1","2","a","4","5"))
	.map(_.toInt)
	.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
	.runForeach(println)
	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]
	Source(1 to 10)
	.via(nonLinearFlow)
	.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
	.runForeach(println)
	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]])