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)

pitfalls_7_1.scala

val infiniteTransactionsStream =
 Stream.continually(s"$randomUser() sent $randomValue() BTC to $randomUser()")

Source(infiniteTransactionsStream)
 .grouped(5) //group transactions into 5 element blocks
 .scan(0.hashCode.toString) { (previousBlockHash, transactions) => //calculate hashes for each block.
   (transactions.hashCode(), previousBlockHash).hashCode.toString //hash for block is defined as hash of transactions hash and previous block hash
 }
.runForeach(savedHash => println(s"saved $savedHash"))

pitfalls_7_2.scala

Source(infiniteTransactionsStream)
 .log("got transaction")

pitfalls_7_3.scala

Source(infiniteTransactionsStream)
 .log("got transaction")
 .grouped(5)
 .log("grouped transactions")
 .scan(0.hashCode.toString) { (previousBlockHash, transactions) => 
   (transactions.hashCode(), previousBlockHash).hashCode.toString 
 }
 .log("hashed block")
 .mapAsync(1)(saveHashAndReturnIt)
 .log("saved block")
.withAttributes(logLevels(onElement = BedugLevel))

pitfalls_7_4.scala

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