Skip to content

Instantly share code, notes, and snippets.

View aludwiko's full-sized avatar

Andrzej Ludwikowski aludwiko

41 followers · 0 following
View GitHub Profile
@aludwiko
aludwiko / pitfalls_6.scala
Last active November 17, 2017 11:24
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
aludwiko / akka_streams_pitfalls_6.scala
Created November 17, 2017 11:27
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
aludwiko / akka_streams_pitfalls_7_1.scala
Last active November 29, 2017 06:04
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) =>
//hash for block is defined as hash of transactions hash and previous block hash
  (transactions.hashCode(), previousBlockHash).hashCode.toString
 }
.mapAsync(1)(saveHashAndReturnIt)
@aludwiko
aludwiko / akka_streams_pitfalls_7_2.scala
Created November 17, 2017 11:34
Source(infiniteTransactionsStream)
 .log(“got transaction”)
@aludwiko
aludwiko / akka_streams_pitfalls_7_3.scala
Last active November 28, 2017 17:50
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”)
@aludwiko
aludwiko / akka_streams_pitfalls_7_4.scala
Created November 17, 2017 11:57
def saveHashAndReturnIt(hash: String): Future[String] = Future {
 if (Random.nextInt(5) == 1) {
  Thread.sleep(100000) //simulate database unavailability
 }
 hash
}
@aludwiko
aludwiko / akka_streams_pitfalls_8.scala
Created November 17, 2017 12:05
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
aludwiko / akka_streams_pitfalls_8_1.scala
Created November 17, 2017 12:07
val roomsTemperatureReadings =
 Stream.continually(TemperatureReading(randomRoom, randomTemperature))
Source(roomsTemperatureReadings)
 .alsoTo(Sink.foreach(reading => createRoomIfNotExist(reading.roomId)))
 .to(Sink.foreach(saveReading))
 .run()
@aludwiko
aludwiko / akka_streams_pitfalls_9.scala
Last active November 17, 2017 12:13
Source(List("1","2","a","4","5"))
 .map(_.toInt)
 .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
 .runForeach(println)
@aludwiko
aludwiko / akka_streams_pitfalls_9_1.scala
Created November 17, 2017 12:16
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]