class: center, middle # Deep Dive Into Scala and Akka ## Yaroslav Tkachenko --- # About me .left-column.column-30[ ![](http://1.gravatar.com/avatar/565c77691f1676e0bab82b4881361f55?size=200) - Java, Scala - Node.js - Python - Microservices - Distributed Systems - DevOps - ... [and more](http://sap1ens.com/files/resume-yaroslav-tkachenko.pdf) ] .right-column.column-70[ .text-center[ **Yaroslav (Slava) Tkachenko**, Software engineer ] .text-right[ Vancouver, BC, Canada ] **Mobify** - Senior Software Engineer, Lead, November 2016 – Present **Bench Accounting** - Director of Engineering, October 2015 – October 2016 - Engineering Lead, April 2014 – September 2015 - Software Engineer, September 2011 – March 2014 **Freelance** - Web Developer, 2007 - 2011 ] ??? Ask everyone for 10 seconds introductions Thank for participation Feel free to interrupt me anytime and ask questions --- # Agenda - Scala basics - Functional programming in Scala - Scala Futures - Actor-based programming: theory, common patterns - Actors in Akka - Supervision - Solving practical problem with Actors and Akka - Akka HTTP - Akka Clustering: sharding and singleton - Akka Persistence with Cassandra - Akka Streams - Testing in Akka ??? Mention Lightbend --- # Why Scala & Akka? .center[ ![](http://sap1ens-archive.s3-website-us-east-1.amazonaws.com/pictures/scala.jpg) ## Scala ] ??? We're not going to learn all language features. Our goal is to know enough to use Akka. --- class: center, middle # Scala basics --- # Scala basics ## Scala - General-purpose - Object-oriented - Functional - Static typing - JVM runtime ## Scala REPL \> scala --- # Scala basics - Primitive types and functions **"Primitive"** types: Byte, Char, Short, Int, Long, Float, Double, Boolean, String... Variables can be **mutable** and **immutable**: `var` and `val` Functions are defined using `def`, for example: `def length(n: String): Int` Hello World! ```scala object HelloWorld { def main(args: Array[String]): Unit = { println("Hello, world!") } } ``` ??? Symbols are interned and can be compared in constant time (O(1)) --- # Scala basics - Conditionals and loops Everything is an expression (that's why we don't need `return`)! ```scala if (x > 0) 1 else if (x == 0) 0 else -1 ``` ```scala for (i <- 1 to n) r = r * i ``` ```scala for (i <- 1 to 3; j <- 1 to 3 if i != j) print((10 * i + j) + " ") ``` ```scala for (i <- 1 to 10) yield i % 3 ``` ```scala var sum = 0 var i = 0 while (i <= 1000 && sum <= 1000) { sum += 1; i += 1 } ``` ??? - No `break` or `continue` operators. There is `import scala.util.control.Breaks._` with `breakable` and `break`, using exceptions. - For comprehensions are translated in series of flatMaps and filters --- # Scala basics - Pattern matching ```scala def matchTest(x: Any, y: Int): String = x match { case 1 => "one" case "two" => "2" case 3 | 4 | 5 => "3, 4, 5" case "ten" if y > 100 => "10" case y: Int => s"Int: $y" case _ => "idk" } ``` ```scala try { process(new URL("http://sap1ens.com")) } catch { case _: MalformedURLException => println("Bad URL: " + url) case ex: IOException => ex.printStackTrace() } ``` --- # Scala basics - Classes ```scala class Point(var x: Int, var y: Int) { def move(dx: Int, dy: Int): Unit = { x = x + dx y = y + dy } override def toString: String = "(" + x + ", " + y + ")" } ``` ```scala case class Point(x: Int, y: Int) // copy // toString // compare // ... def findX(p: Point): Int = p match { case Point(x, _) => x } ``` --- # Scala basics - Objects ```scala object Helper { def doSomething(thing: String): String = { /* ... */ } } ``` ```scala class Event(var metadata: String) object Event { def apply(): Event = new Event("") def apply(metadata: String): Event = new Event(metadata) } ``` ??? Case classes behave like they have `apply` methods defined --- # Scala basics - Traits ```scala trait One { def one(): String } trait Two { def two(): String = "two" } class Three extends One with Two { def one(): String = "one" } ``` ```scala sealed trait Account case object Chequing extends Account case object Savings extends Account case class CreditCard(user: String) extends Account ``` --- # Scala basics - Generics ```scala class Stack[T] { var elems: List[T] = Nil def push(x: T) { elems = x :: elems } def top: T = elems.head def pop() { elems = elems.tail } } ``` ```scala def count[A](l: List[A]) = l.size def count(l: List[_]) = l.size // "wildcard" ``` ```scala class Box[F <: Fruit](aFruit: F) { def fruit: F = aFruit def contains(aFruit: Fruit) = fruit.name == aFruit.name } val appleBox = new Box[Apple](new Apple) val orangeBox = new Box[Orange](new Orange) ``` --- # Scala basics - Collections .center[ ![](http://docs.scala-lang.org/resources/images/collections.png) ] --- # Scala basics - Collections ```scala val list = List(1, 2, 3) val map = HashMap("x" -> 24, "y" -> 25, "z" -> 26) val set = Set(Color.red, Color.green, Color.blue) val tuple = (1, "localhost") ``` - `foreach` - `map` - `filter` - `find` - `drop` / `dropWhile` - `foldRight` / `foldLeft` - `flatten` - `flatMap` - `count` - `groupBy` - ... - by the way, you can access sequences as arrays using index (starting from 0), for example `list(0)` --- # Scala basics - Collections ```scala List(1, 2, 3).foreach(element => println(element)) ``` ```scala List(1, 2, 3) foreach { element => println(element) } ``` ```scala List(1, 2, 3).foreach(println(_)) ``` ```scala List(1, 2, 3).foreach { println(_) } ``` ```scala List(1, 2, 3).foreach { println _ } ``` ```scala List(1, 2, 3).foreach(println) ``` ```scala List(1, 2, 3) foreach (println) ``` ```scala List(1, 2, 3) foreach { println } ``` ```scala List(1, 2, 3) foreach println ``` --- # Scala basics - Packages and imports ``` - com - sap1ens - data - App.scala ``` ```scala package com.sap1ens.data object Thing { def doSomething(): Unit } ``` ```scala import com.sap1ens.data.Thing Thing.doSomething() ``` --- # Scala basics - Implicits ## Implict parameters ```scala // probably in a library class Prefixer(val prefix: String) def addPrefix(s: String)(implicit p: Prefixer) = p.prefix + s // then probably in your application implicit val myImplicitPrefixer = new Prefixer("***") addPrefix("abc") // returns "***abc" ``` ## Implicit conversions ```scala implicit def doubleToInt(d: Double) = d.toInt val x: Int = 42.0 ``` --- class: center, middle # It's time to code! --- # It's time to code! Task: - Download [this file](https://raw.githubusercontent.com/jokecamp/FootballData/master/Euro%202012/Euro%202012%20stats%20TEAM.csv) - Create a case class to represent a team - Parse CSV file and keep results in memory. Only use the columns you need (heads up - you'll need 0, 1 and 12) - Find: - Team with a biggest number of goals - Teams starting with 'G' or 'P' characters and more than 5 goals - Team that completed 1200 passes. Print its name if team exists, otherwise print "Nothing found" Tips: - Try to only use Scala Collections API methods - Remember how to access elements of arrays using indexes - Get lines from a file: ```scala val lines = scala.io.Source.fromFile("file.txt").getLines ``` --- # It's time to code! ```scala import scala.io.Source case class Team(name: String, goals: Int, passes: Int) val lines = Source.fromFile("/Users/sap1ens/Downloads/euro-stats.csv").getLines.toList.tail val teams: List[Team] = for(line <- lines) yield { val teamRaw = line.split(",") Team(teamRaw(0), teamRaw(1).toInt, teamRaw(12).toInt) } val teamWithBiggestGoals: Team = teams.maxBy(_.goals) val teamsStartingWithGP: List[Team] = teams .filter(team => team.name.startsWith("G") || team.name.startsWith("P")) .filter(_.goals > 5) val teamWith1200CompletedPasses: Option[Team] = teams.find(_.passes == 1200) teamWith1200CompletedPasses match { case Some(team) => println(team.name) case None => println("Nothing found") } ``` --- class: center, middle # Functional programming in Scala --- # Functional programming in Scala .center.middle[ http://www.lihaoyi.com/post/WhatsFunctionalProgrammingAllAbout.html ] --- # Functional programming in Scala Imperative recipe: ```python def make_tiramisu(eggs, sugar1, wine, cheese, cream, fingers, espresso, sugar2, cocoa): dissolve(sugar2, espresso) mixture = whisk(eggs) beat(mixture, sugar1, wine) whisk(mixture) # over steam whip(cream) beat(cheese) beat(mixture, cheese) fold(mixture, cream) assemble(mixture, fingers) sift(mixture, cocoa) refrigerate(mixture) return mixture # it's now a tiramisu ``` --- # Functional programming in Scala Functional recipe: ```python def make_tiramisu(eggs, sugar1, wine, cheese, cream, fingers, espresso, sugar2, cocoa): beat_eggs = beat(eggs) mixture = beat(beat_eggs, sugar1, wine) whisked = whisk(mixture) beat_cheese = beat(cheese) cheese_mixture = beat(whisked, beat_cheese) whipped_cream = whip(cream) folded_mixture = fold(cheese_mixture, whipped_cream) sweet_espresso = dissolve(sugar2, espresso) wet_fingers = soak2seconds(fingers, sweet_espresso) assembled = assemble(folded_mixture, wet_fingers) complete = sift(assembled, cocoa) ready_tiramisu = refrigerate(complete) return ready_tiramisu ``` ??? - No side-effects - Clear inputs and outputs - Easy to refactor and test - Easy to see dependencies (chain of calls) and missing variables --- # Functional programming in Scala .center[ ![](http://sap1ens-archive.s3-website-us-east-1.amazonaws.com/pictures/TiramisuDiagram.png) ![](http://sap1ens-archive.s3-website-us-east-1.amazonaws.com/pictures/DiagramGraph.png) ] --- # Functional programming in Scala - Functions are first class objects - Collections: ```scala def filter(p: Int => Boolean): List[Int] ``` - Immutability, laziness, tail recursion - Immutable collections, case classes, proper data types and algorithms - Lazy vals: ```scala lazy val isReady = expensiveFunc() ``` - Tail recursion ```scala @tailrec ``` - Types (*Category theory*) --- # Functional programming in Scala - Option type - Ever seen `NullPointerException`? - How about writing `foo?.bar?.baz`? - And ```javascript if(test.level1 && test.level1.level2 && test.level1.level2.level3) { // ... } ``` ? Scala's solution: ```scala sealed trait Option case object None extends Option case class Some(value: Any) extends Option ``` ```scala val absentGreeting: Option[String] = Option(null) // absentGreeting will be None val presentGreeting: Option[String] = Option("Hello!") // presentGreeting will be Some("Hello!") ``` --- # Functional programming in Scala - Option type ```scala // 1 optionVar match { case Some(i) => println(i) case None => println("That didn't work.") } // 2 if(optionVar.isDefined) { println(optionVar.get) } else { println("That didn't work.") } // 3 optionVar map { i => println(i) } getOrElse { println("That didn't work.") } ``` ??? - Missing error message / details in case of None (use Either) --- # Functional programming in Scala - Either type `Option` is great, but I also want to pass error message / details! Scala's solution: ```scala sealed trait Either case class Left(value: Any) extends Either case class Right(value: Any) extends Either ``` ```scala def getContent(url: URL): Either[String, Source] = if (url.getHost.contains("google")) Left("Requested URL is blocked for the good of the people!") else Right(Source.fromURL(url)) ``` --- # Functional programming in Scala - Either type Either is *unbiased* ```scala // 1 val content: Either[String, Iterator[String]] = getContent(new URL("http://sap1ens.com")).right.map(_.getLines()) // 2 validateName(inputName).fold { error => s"Validation failed: $error", result => s"Validation succeeded: $result" } // 3 validateName(inputName) match { case Left(error) => s"Validation failed: $error", case Right(result) => s"Validation succeeded: $result" } getContent(new URL("http://sap1ens.com")).right.toOption ``` --- class: center, middle # Scala Futures --- # Scala Futures ```scala object Future { def apply[T](body: => T)(implicit execctx: ExecutionContext): Future[T] } ``` ```scala import scala.concurrent.Future import scala.concurrent.ExecutionContext.Implicits.global import scala.util.{Success, Failure} def fetchURL(url: String): Future[String] = Future { scala.io.Source.fromURL(url).getLines.mkString } // 1 fetchURL("http://sap1ens.com") onComplete { case Success(body) => println(body) case Failure(ex) => println("Sorry, couldn't fetch that") } // 2 fetchURL("http://sap1ens.com") map { body => println(body) } recover { case ex => println("Sorry, couldn't fetch that") } ``` --- # Scala Futures ```scala Future.sequence(Seq(future1, future2, future3)) ``` ```scala val first = Future.successful("yeah") val second = Future.failed(new Exception("nope")) ``` ```scala import scala.concurrent._ import scala.concurrent.duration._ val request = fetchURL("http://sap1ens.com") Await.result(request, 5 seconds) ``` ```scala val usdQuote = Future { connection.getCurrentValue(USD) } val chfQuote = Future { connection.getCurrentValue(CHF) } val purchase = for { usd <- usdQuote chf <- chfQuote if isProfitable(usd, chf) } yield connection.buy(amount, chf) purchase onSuccess { case _ => println("Purchased " + amount + " CHF") } ``` --- # SBT - Scala Build Tool http://www.scala-sbt.org/download.html or just `brew install sbt`. Maven-like folder structure (created automatically by Intellij IDEA): ``` - project - plugins.sbt - src - main - resources - scala - com.sap1ens - Service.scala - test - resources - scala - com.sap1ens - build.sbt ``` Service.scala: ```scala object Service extends App { // put some code here... } ``` --- # SBT - Scala Build Tool build.sbt: ```scala name := """example-service""" version := "1.0" scalaVersion := "2.12.1" libraryDependencies ++= Seq( "com.typesafe" % "config" % "1.3.1" ) ``` Commands: - `sbt run` - `sbt console` - `sbt test` --- # Typesafe config https://github.com/typesafehub/config `application.conf` is a standard config file, should be located in `src/main/resources`: ``` akka { loggers = ["akka.event.slf4j.Slf4jLogger"] loglevel = "DEBUG" } db.connection_string = ${MYSQL_CONNECTION_STRING} hostname = "<IP_ADDRESS>" port = 80 ``` Use it like this: ```scala val config = ConfigFactory.load() // it's a good idea to implement it like a trait val port = conf.getInt("port") ``` --- class: center, middle # Actor systems --- # Actor systems An actor is a computational entity that, in response to a message it receives, can concurrently: - send a finite number of messages to other actors; - create a finite number of new actors; - designate the behavior to be used for the next message it receives. There is no assumed sequence to the above actions and they could be carried out in parallel. Key concepts: - Actor - Actor systems - Mailbox - Supervision strategy --- # Actor systems - Example .left-column.column-50.text-bigger[ 📬 **Rachel** ] .right-column.column-50.text-right.text-bigger[ 📬 **Alex** ] .center.text-bigger[ 📬 **Fred** ] --- # Actor systems - Supervision .center.text-bigger[ 📬 **Rachel** ] .center.text-bigger[ 💰 ⬇ ] .center.text-bigger[ 📬 **Fred** ] --- # Actor systems - Supervision .center.text-bigger[ 📬 **Rachel** ] .center.text-bigger[ 💰 ⬇ ] .center.text-bigger[ 📬 **Fred** ❌ ] --- # Actor systems - Supervision .center.text-bigger[ 📬 **Rachel** ] .center.text-bigger[ 💰 ⬇ ⬆ ] .center.text-bigger[ 📬 **Fred** ❌ ] --- # Actor systems - Supervision Strategies: - Resume the subordinate, keeping its accumulated internal state - Restart the subordinate, clearing out its accumulated internal state - Stop the subordinate permanently - Escalate the failure, thereby failing itself --- class: center, middle # Akka --- # Akka Akka is an open-source toolkit and runtime simplifying the construction of concurrent and distributed applications on the JVM. Akka supports multiple programming models for concurrency, but it emphasizes actor-based concurrency, with inspiration drawn from Erlang. --- # Akka - Overview Why? - Concurrency and overall performance - Scalability - Clean asynchronous programming model - Resilience - Simple distributed programming --- # Akka - Reactive Manifesto .center[ ![](http://www.reactivemanifesto.org/images/reactive-traits.svg) ] --- # Akka - Concurrency Scripting languages like PHP, Ruby, Python – usually single-threaded C, C#, Java – explicit thread management Erlang, Haskell, Go – “green threads”, channels, etc. Scala: Akka – actors (“green threads”) + Java Virtual Machine --- # Akka - Actors ```scala case class Greeting(who: String) class GreetingActor extends Actor with ActorLogging { def receive = { case Greeting(who) => log.info("Hello " + who) } } val system = ActorSystem("MySystem") val greeter = system.actorOf(Props[GreetingActor], name = "greeter") greeter ! Greeting("Charlie Parker") ``` - Immutability of messages - Messages are handled one by one - Stateful actors --- class: center, middle ![](http://doc.akka.io/docs/akka/2.4/_images/actor_lifecycle1.png) --- # Akka - Actors Hooks: ```scala def preStart() def postStop() def preRestart(reason: Throwable, message: Option[Any]) def postRestart(reason: Throwable) ``` --- # Akka - Props A proper way to create Actors with parameters: ```scala object DemoActor { def props(magicNumber: Int): Props = Props(classOf[DemoActor], magicNumber) } class DemoActor(magicNumber: Int) extends Actor with ActorLogging { def receive = { case x: Int => log.info(x + magicNumber) } } val demoActor = context.actorOf(DemoActor.props(42), "demo") ``` --- # Akka - Communication ```scala class HelloActor extends Actor with ActorLogging { def receive = { case who => sender() ! "Hello, " + who } } object ConversationActor { def props(fellowActor: ActorRef): Props = Props(classOf[ConversationActor], fellowActor) } class ConversationActor(fellowActor: ActorRef) extends Actor with ActorLogging { def receive = { case "start" => fellowActor ! "it's me!" case message => log.info(message) } } val system = ActorSystem("MySystem") val helloActor = system.actorOf(Props[HelloActor]) val conversationActor = ConversationActor.props(helloActor) conversationActor ! "start" ``` --- # Akka - Communication Ask: ```scala import scala.concurrent.duration._ import akka.util.Timeout import akka.pattern.ask implicit val timeout = Timeout(5 seconds) val future = myActor ? "hello" ``` Forward: ```scala target forward message ``` Kill: ```scala import akka.actor.PoisonPill target ! PoisonPill ``` Watch: ```scala actor watch target ``` ??? `ask` doesn't allow to receive messages about actor termination, `watch` is used instead. --- # Akka - Behavior ```scala class HotSwapActor extends Actor { import context._ def angry: Receive = { case "foo" => sender() ! "I am already angry?" case "bar" => become(happy) } def happy: Receive = { case "bar" => sender() ! "I am already happy :-)" case "foo" => become(angry) } def receive = { case "foo" => become(angry) case "bar" => become(happy) } } ``` --- # Akka - Behavior ```scala import akka.actor.Stash class ActorWithProtocol extends Actor with Stash { def receive = { case "open" => unstashAll() context.become({ case "write" => // do writing... case "close" => unstashAll() context.unbecome() case msg => stash() }, discardOld = false) // stack on top instead of replacing case msg => stash() } } ``` --- # Akka - Routing ```scala val router: ActorRef = context.actorOf(RoundRobinPool(5).props(Props[Worker]), "router") ``` - RoundRobinRoutingLogic - RandomRoutingLogic - SmallestMailboxRoutingLogic - BroadcastRoutingLogic - ScatterGatherFirstCompletedRoutingLogic - TailChoppingRoutingLogic - ConsistentHashingRoutingLogic --- # Akka - Mailboxes ``` control-aware-dispatcher { mailbox-type = "akka.dispatch.UnboundedControlAwareMailbox" } ``` ```scala val a = system.actorOf(Props(classOf[Logger], this).withDispatcher("control-aware-dispatcher")) ``` - **UnboundedMailbox** - SingleConsumerOnlyUnboundedMailbox - NonBlockingBoundedMailbox - UnboundedControlAwareMailbox - UnboundedPriorityMailbox - UnboundedStablePriorityMailbox - BoundedMailbox - BoundedPriorityMailbox - BoundedStablePriorityMailbox - BoundedControlAwareMailbox --- # Akka - Event Stream ```scala system.eventStream.subscribe(listenerActor, classOf[Message]) ``` ```scala system.eventStream.publish(Message("Hi!")) ``` For example: ```scala import akka.actor.AllDeadLetters system.eventStream.subscribe(listener, classOf[AllDeadLetters]) ``` or local Pub/Sub. --- # Akka - Supervision ```scala class Supervisor extends Actor { override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) { case _: ArithmeticException ⇒ Resume case _: NullPointerException ⇒ Restart case _: Exception ⇒ Escalate } val worker = context.actorOf(Props[Worker]) def receive = { case n: Int => worker forward n } } ``` --- # Akka - Actors & Futures ```scala import akka.pattern.{ask, pipe} import system.dispatcher // The ExecutionContext that will be used final case class Result(x: Int, s: String, d: Double) case object Request implicit val timeout = Timeout(5 seconds) // needed for `?` below val f: Future[Result] = for { x <- ask(actorA, Request).mapTo[Int] // call pattern directly s <- (actorB ask Request).mapTo[String] // call by implicit conversion d <- (actorC ? Request).mapTo[Double] // call by symbolic name } yield Result(x, s, d) f pipeTo actorD // .. or .. pipe(f) to actorD ``` --- class: center, middle # Blocking Is Bad http://stackoverflow.com/questions/13097754/asynchronous-io-in-scala-with-futures http://stackoverflow.com/questions/19681389/use-case-of-scala-concurrent-blocking --- # Akka and Application Architecture *We have actors. What are supposed to do with them? MVC? SOA? Something else? How about DI? How do we structure actor-based apps?* Let's talk about: - **DDD** (Domain-Driven Design) - **EIP** (Enterprise Integration Patterns) --- # Akka and Application Architecture .left-column.column-45.text-right[ ![](https://images-na.ssl-images-amazon.com/images/I/5146azDZjmL._SX358_BO1,204,203,200_.jpg) ] .right-column.column-45[ ![](https://images-na.ssl-images-amazon.com/images/I/51dGyfBYvuL._SX382_BO1,204,203,200_.jpg) ] --- # Akka and Application Architecture .left-column.column-50[ Domain Driven Design: - Value Object - Entity - Service - Repository - Domain Event - Event vs Command - ... ] .right-column.column-50[ Enterprise Integration Patterns: - Message - Point-to-Point - Publish-Subscribe - Pipes and Filters - Content-Based Router - Message Filter - Scatter-Gather - ... ] --- # Akka and Application Architecture .center[ ![](http://sap1ens-archive.s3-website-us-east-1.amazonaws.com/pictures/workshop_akka_design_example.png) ] --- # Akka and Application Architecture ```scala object SynchronizationService { // data model sealed trait DocumentMetadata { def version: String } case class DocumentMetadataV2(...) extends DocumentMetadata case class DocumentMetadataV3(...) extends DocumentMetadata // actions case class SaveDocument(...) case class StartSync(connection: Connection) private case class CreateStatement(...) private case class FindOrCreateStatementAccount(...) private case class CreateStatementAccount(...) private case class SaveAndUploadStatement(...) // ... ``` --- # Akka and Application Architecture ```scala // ... // results sealed trait DeliveryResult case object Delivered extends DeliveryResult case object Failed extends DeliveryResult case object NotFound extends DeliveryResult private case class SyncFailure(...) def props(s3Service: ActorRef, exportService: ActorRef, connectionsResource: ConnectionsResourceLike, filestorePath: String): Props = Props(classOf[SynchronizationService], s3Service, exportService, connectionsResource, filestorePath) } ``` --- # Akka and Application Architecture ```scala class SynchronizationService(s3Service: ActorRef, exportService: ActorRef, connectionsResource: ConnectionsResourceLike, filestorePath: String) extends Actor with ActorLogging { def receive: Receive = { case StartSync(connection) => // ... case SaveDocument(metadata, file) => // ... case action @ CreateStatement(...) => // ... case FindOrCreateStatementAccount(action, connection, receiver) => // ... case CreateStatementAccount(action, connection, receiver) => // ... case SaveAndUploadStatement(action, connection, statementAccount, receiver) => // ... case SyncFailure(error, action, receiver, deliveryResult) => // ... case UploadResultOk(statement, receiver) => // ... case UploadResultFailure(error, receiver) => // ... } } ``` --- # Akka and Application Architecture ```scala trait CoreActors extends ConfigHolder { // helpers val integrationApiClient = new APIClient() val s3Service = system.actorOf(S3Service.props(config.getString("s3.bucket"), config.getString("s3.accessKey"), config.getString("s3.secretKey"), config.getInt("s3.maxRetries") ), "S3Service") val fileManager = system.actorOf(FileManager.props(config.getString("filestore"), s3Service), "FileManager") // services val exportService = system.actorOf(ExportService.props( integrationApiClient, fileManager, Duration(config.getInt("export.interval"), MINUTES), config.getBoolean("export.enabled"), config.getInt("export.maxRetries") ), "StatementsExportService") val synchronizationService = system.actorOf(SynchronizationService.props( s3Service, exportService, ConnectionsResource, config.getString("filestore") ), "SynchronizationService") } ``` --- # Akka and Application Architecture We've just created Application **Core**. We can add more interfaces like **HTTP** and **message queues**. - HTTP, we'll look into Akka HTTP soon - Message queues, let's use Akka Camel --- # ~~Akka~~ Apache Camel http://camel.apache.org "The most unknown coolest library out there" JM (c) *Apache Camel offers you the interfaces for the EIPs, the base objects, commonly needed implementations, debugging tools, a configuration system, and many other helpers which will save you a ton of time when you want to implement your solution to follow the EIPs.* ```java from("direct:report") .to("file:target/reports/?fileName=report.txt") from("twitter://search?...") .to("websocket:camel-tweet?sendToAll=true") from("netty-http:http://<IP_ADDRESS>:8080") .to("direct:name") from("jms:invoices") .setBody() .groovy("new Invoice(request.body,currentTimeMillis())") .to("mongodb:mongo?...operation=insert") ``` Hundreds of components for everything! https://camel.apache.org/components.html --- # Akka Camel Akka Camel Consumer: ```scala class OrdersConsumer extends Consumer { def endpointUri = "activemq:queue:orders" def receive = { case msg: CamelMessage => { /* ... */ } case _ => { /* ... */ } } } ``` Akka Camel Producer: ```scala class OrdersProducer extends Actor with Producer { def endpointUri = "activemq:queue:orders" } val orders = system.actorOf(Props[Orders]) orders ! Order(...) ``` --- # Akka and Application Architecture General guidelines: - Set of Actors as a core layer - Additional layers for integration, persistence, etc. - Akka HTTP for REST API - Akka Camel for message queues --- # Akka and Application Architecture Hexagonal architecture: .center[ ![](http://alistair.cockburn.us/get/2301) ] --- # Akka and Application Architecture Hexagonal architecture: .center[ ![](http://alistair.cockburn.us/get/2304) ] --- class: center, middle # It's time to code, again! --- # It's time to code, again! Task: https://github.com/sap1ens/scala-akka-workshop (use this to start). Let's build a [job queue](https://en.wikipedia.org/wiki/Job_queue)! In our case we're going to have `JobExecutor` actor that accepts incoming job requests and executes them using pool of `JobWorker` actors. If one of the `JobWorker` actors fails `JobExecutor` should restart failed job. Also `JobWorker`s should communicate results to `JobExecutor`. A job itself can be anything (read or write a file, calculate something, etc.), but I suggest to download a file from specified URL and save it to local disk with provided path. Ideally we should introduce a percent of failures, so we can see restarts as well. A few important things: - All Akka dependencies are already included in build.sbt - You can run the application with `sbt run` - Check `SimpleActor` and `Helpers` files - Use case classes to represent jobs - You will probably need to use routing --- # It's time to code, again! Solution: https://github.com/sap1ens/scala-akka-workshop/tree/solution/src/main/scala/com/sap1ens/workshop Solution with Futures: https://github.com/sap1ens/scala-akka-workshop/tree/solution-with-futures/src/main/scala/com/sap1ens/workshop --- class: center, middle # Akka Streams --- # Akka Streams http://www.reactive-streams.org Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure. This encompasses efforts aimed at runtime environments (JVM and JavaScript) as well as network protocols. In summary, Reactive Streams is a standard and specification for Stream-oriented libraries for the JVM that - process a potentially unbounded number of elements - in sequence, - asynchronously passing elements between components, - with mandatory non-blocking backpressure. --- # Akka Streams - Source - Flow - Sink and also Graphs! ```scala implicit val actorSystem = ActorSystem("akka-streams-example") implicit val materializer = ActorMaterializer() val helloWorldStream: RunnableGraph[NotUsed] = Source.single("Hello world") .via(Flow[String].map(s => s.toUpperCase())) .to(Sink.foreach(println)) helloWorldStream.run() ``` --- # Akka Streams .left-column.column-33[ Source stages: - fromIterator - apply - single - repeat - cycle - tick - fromFuture - fromCompletionStage - unfold - unfoldAsync - empty - maybe - ... ] .left-column.column-33[ Simple Flow processing stages: - map - mapConcat - statefulMapConcat - filter - filterNot - collect - grouped - sliding - scan - scanAsync - fold - foldAsync - ... ] .left-column.column-33[ Sink stages: - head - headOption - last - lastOption - ignore - cancelled - seq - foreach - foreachParallel - onComplete - lazyInit - queue - ... ] --- # Akka Streams ```scala implicit val system = ActorSystem("Sys") implicit val materializer = ActorMaterializer() val LoglevelPattern = """.*\[(DEBUG|INFO|WARN|ERROR)\].*""".r FileIO.fromPath(Paths.get("src/main/resources/logfile.txt")) // parse chunks of bytes into lines .via(Framing.delimiter(ByteString(System.lineSeparator), maximumFrameLength = 512, allowTruncation = true)) .map(_.utf8String) .map { case line @ LoglevelPattern(level) => (level, line) case line @ other => ("OTHER", line) } .groupBy(5, _._1) // group them by log level .fold(("", List.empty[String])) { case ((_, list), (level, line)) => (level, line :: list) } .mapAsync(parallelism = 5) { // write lines of each group to a separate file case (level, groupList) => Source(groupList.reverse) .map(line => ByteString(line + "\n")).runWith(FileIO.toPath(Paths.get(s"target/log-$level.txt"))) } .mergeSubstreams .runWith(Sink.onComplete { case Success(_) => system.terminate() case Failure(e) => println(s"Failure: ${e.getMessage}") system.terminate() }) ``` --- # Akka Streams ```scala val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] => import GraphDSL.Implicits._ val in = Source(1 to 10) val out = Sink.ignore val bcast = builder.add(Broadcast[Int](2)) val merge = builder.add(Merge[Int](2)) val f1, f2, f3, f4 = Flow[Int].map(_ + 10) in ~> f1 ~> bcast ~> f2 ~> merge ~> f3 ~> out bcast ~> f4 ~> merge ClosedShape }) ``` --- class: center, middle # Akka HTTP --- # Akka HTTP - Reincarnation of the famous [Spray](http://spray.io) library - Akka-native (based on Akka Streams) - Very fast and performant - Very strict implementation - Has Server and Client APIs - Server APIs: - Low-level (dealing with connections and streams) - High-level (using routing and directives) - Client APIs: - Low-level Connection API - Low-level Host API - High-level Request API --- # Akka HTTP - High-level Server Example ```scala import akka.actor.ActorSystem import akka.http.scaladsl.Http import akka.http.scaladsl.model._ import akka.http.scaladsl.server.Directives._ import akka.stream.ActorMaterializer object WebServer { def main(args: Array[String]) { implicit val system = ActorSystem("my-system") implicit val materializer = ActorMaterializer() val route = path("hello") { get { complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "
Say hello to akka-http
")) } } Http().bindAndHandle(route, "localhost", 8080) } } ``` --- # Akka HTTP - More Routes ```scala val route = { path("orders") { authenticateBasic(realm = "admin area", myAuthenticator) { user => get { encodeResponseWith(Deflate) { complete { // marshal custom object with in-scope marshaller retrieveOrdersFromDB } } } ~ post { // decompress gzipped or deflated requests if required decodeRequest { // unmarshal with in-scope unmarshaller entity(as[Order]) { order => complete { // ... write order to DB "Order received" } } } } } } ``` --- # Akka HTTP - More Routes ```scala val route = { // extract URI path element as Int pathPrefix("order" / IntNumber) { orderId => pathEnd { put { // form extraction from multipart or www-url-encoded forms formFields(('email, 'total.as[Money])).as(Order) { order => complete { // complete with serialized Future result (myDbActor ? Update(order)).mapTo[TransactionResult] } } } } ~ path("items") { get { // parameters to case class extraction parameters(('size.as[Int], 'color ?, 'dangerous ? "no")) .as(OrderItem) { orderItem => // ... route using case class instance created from // required and optional query parameters complete("") // hide } } } } ``` --- # Akka HTTP - More Routes ```scala val route = { pathPrefix("documentation") { // optionally compresses the response with Gzip or Deflate // if the client accepts compressed responses encodeResponse { // serve up static content from a JAR resource getFromResourceDirectory("docs") } } ~ path("oldApi" / Remaining) { pathRest => redirect("http://oldapi.example.com/" + pathRest, MovedPermanently) } } ``` --- # Akka HTTP - The Routing Tree ```scala val route = a { b { c { ... // route 1 } ~ d { ... // route 2 } ~ ... // route 3 } ~ e { ... // route 4 } } ``` - Route 1 will only be reached if directives `a`, `b` and `c` all let the request pass through. - Route 2 will run if `a` and `b` pass, `c` rejects and `d` passes. - Route 3 will run if `a` and `b` pass, but `c` and `d` reject. --- # Akka HTTP - Serialization ```scala import akka.http.scaladsl.server.Directives import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport import spray.json._ final case class Item(name: String, id: xrLong) final case class Order(items: List[Item]) trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol { implicit val itemFormat = jsonFormat2(Item) implicit val orderFormat = jsonFormat1(Order) // contains List[Item] } class MyJsonService extends Directives with JsonSupport { val route = get { pathSingleSlash { complete(Item("thing", 42)) // will render as JSON } } ~ post { entity(as[Order]) { order => // will unmarshal JSON to Order val itemsCount = order.items.size val itemNames = order.items.map(_.name).mkString(", ") complete(s"Ordered $itemsCount items: $itemNames") } } } ``` --- # Akka HTTP - Low-level Server Example ```scala implicit val system = ActorSystem() implicit val materializer = ActorMaterializer() implicit val executionContext = system.dispatcher val serverSource = Http().bind(interface = "localhost", port = 8080) val requestHandler: HttpRequest => HttpResponse = { case HttpRequest(GET, Uri.Path("/"), _, _, _) => HttpResponse(entity = HttpEntity(ContentTypes.`text/html(UTF-8)`, "Hello world!")) case HttpRequest(GET, Uri.Path("/ping"), _, _, _) => HttpResponse(entity = "PONG!") case HttpRequest(GET, Uri.Path("/crash"), _, _, _) => sys.error("BOOM!") case r: HttpRequest => r.discardEntityBytes() // important to drain incoming HTTP Entity stream HttpResponse(404, entity = "Unknown resource!") } val bindingFuture: Future[Http.ServerBinding] = serverSource.to(Sink.foreach { connection => println("Accepted new connection from " + connection.remoteAddress) connection handleWithSyncHandler requestHandler }).run() ``` --- # Akka HTTP / Streams Akka HTTP is streaming *all the way through*, which means that the back-pressure mechanisms enabled by Akka Streams are exposed through all layers–from the TCP layer, through the HTTP server, all the way up to the user-facing `HttpRequest` and `HttpResponse` and their `HttpEntity` APIs. This has surprising implications if you are used to non-streaming / not-reactive HTTP clients. Specifically it means that: “*lack of consumption of the HTTP Entity, is signaled as back-pressure to the other side of the connection*”. This is a feature, as it allows one only to consume the entity, and back-pressure servers/clients from overwhelming our application, possibly causing un-necessary buffering of the entity in memory. --- class: center, middle # Akka Persistence --- # Akka Persistence - Overview .center[ ![](http://sap1ens-archive.s3-website-us-east-1.amazonaws.com/pictures/akka-persistence.jpg) ] --- # Akka Persistence - Overview - Event Sourcing - Journal - Persistent Actor - Persistent View Has plugins for JDBC (MySQL, Postgres, ...), MongoDB, Cassandra, Kafka, Redis and more. --- # Akka Persistence - Example ```scala case class Cmd(data: String) case class Evt(data: String) case class ExampleState(events: List[String] = Nil) { def updated(evt: Evt): ExampleState = copy(evt.data :: events) override def toString: String = events.reverse.toString } class ExamplePersistentActor extends PersistentActor { override def persistenceId = "sample-id-1" var state = ExampleState() def updateState(event: Evt): Unit = state = state.updated(event) val receiveRecover: Receive = { case evt: Evt => updateState(evt) case SnapshotOffer(_, snapshot: ExampleState) => state = snapshot } val receiveCommand: Receive = { case Cmd(data) => persist(Evt(data))(updateState) case "snap" => saveSnapshot(state) case "print" => println(state) } } ``` --- # Akka Persistence - Example ```scala class MyView extends PersistentView { override def persistenceId: String = "some-persistence-id" override def viewId: String = "some-persistence-id-view" def receive: Receive = { case payload if isPersistent => // handle message from journal... case payload => // handle message from user-land... } } ``` --- # Akka Persistence - Configuration ```scala libraryDependencies += "com.typesafe.akka" %% "akka-persistence-cassandra" % "0.24" ``` ``` akka { persistence { journal.plugin = "cassandra-journal" snapshot-store.plugin = "cassandra-snapshot-store" } } cassandra-journal { contact-points = ["<IP_ADDRESS>:9042, ..."] } ``` More: https://github.com/akka/akka-persistence-cassandra Why Cassandra? - https://github.com/boldradius/akka-dddd-template - http://downloads.typesafe.com/website/presentations/ScalaDaysSF2015/T3_Nash_Easy_scalability.pdf ??? Talk about deployment: 3 akka cluster nodes with persistence, 3 cassandra nodes - all on the same machinces. Interesting consistency model (LOCAL_ONE), but can be higher. --- class: center, middle # Akka Clustering --- # Akka Clustering - Overview .center[ ![](http://sap1ens-archive.s3-website-us-east-1.amazonaws.com/pictures/node-ring.png) ] .left-column.column-50[ - Cluster - Node - Leader ] .right-column.column-50[ - Gossip protocol - Seed nodes - Failure Detector ] --- # Akka Clustering - Overview .center[ ![](http://doc.akka.io/docs/akka/2.4/_images/member-states.png) ] --- # Akka Clustering - Setup ``` akka { actor { provider = cluster } remote { netty.tcp { hostname = "<IP_ADDRESS>" port = 0 } } cluster { seed-nodes = [ "akka.tcp://ClusterSystem@<IP_ADDRESS>:2551", "akka.tcp://ClusterSystem@<IP_ADDRESS>:2552" ] } } ``` --- # Akka Clustering - Setup ```scala package sample.cluster.simple import akka.cluster.Cluster import akka.cluster.ClusterEvent._ import akka.actor.ActorLogging import akka.actor.Actor class SimpleClusterListener extends Actor with ActorLogging { val cluster = Cluster(context.system) override def preStart(): Unit = cluster.subscribe(self, InitialStateAsEvents, classOf[MemberEvent]) override def postStop(): Unit = cluster.unsubscribe(self) def receive = { case MemberUp(member) => log.info("Member is Up: {}", member.address) case MemberRemoved(member, previousStatus) => log.info("Member is Removed: {} after {}", member.address, previousStatus) case _: MemberEvent => } } ``` --- # Akka Clustering - Setup Real world example: https://github.com/sap1ens/akka-cluster-consul ```scala val cluster = Cluster(system) val selfAddress = cluster.selfAddress val serviceAddresses = ConsulAPI.getServiceAddresses // http://doc.akka.io/docs/akka/2.4/scala/cluster-usage.html // // When using joinSeedNodes you should not include the node itself except for the node // that is supposed to be the first seed node, and that should be placed first // in parameter to joinSeedNodes. val serviceSeeds = serviceAddresses filter { address => address != selfAddress || address == serviceAddresses.head } cluster.joinSeedNodes(serviceSeeds) cluster registerOnMemberUp { // init system } ``` --- # Akka Clustering - Actors 3 nodes, 3 actors, but 9 messages: ``` INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9102] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9101] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9101] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9102] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9103] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9103] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9101] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9101] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9103] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9102] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9102] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9103] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9101] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9103] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9102] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9102] INFO - Cluster Node [akka.tcp://demo-system@<IP_ADDRESS>:9103] - Member is Up [akka.tcp://demo-system@<IP_ADDRESS>:9101] ``` --- # Akka Clustering - Singleton Why? - Single point of responsibility for certain cluster-wide consistent decisions, or coordination of actions across the cluster system - Single entry point to an external system - Single master, many workers - Centralized naming service, or routing logic But: - Cluster singleton may quickly become a performance bottleneck - You can not rely on the cluster singleton to be non-stop available — e.g. when the node on which the singleton has been running dies, it will take a few seconds for this to be noticed and the singleton be migrated to another node - In the case of a network partition appearing in a Cluster that is using Automatic Downing (see Auto Downing docs for Downing), it may happen that the isolated clusters each decide to spin up their own singleton, meaning that there might be multiple singletons running in the system, yet the Clusters have no way of finding out about them (because of the partition) --- # Akka Clustering - Singleton Create Singleton: ```scala system.actorOf( ClusterSingletonManager.props( singletonProps = Props(classOf[Consumer], queue, testActor), terminationMessage = End, settings = ClusterSingletonManagerSettings(system)), name = "consumer") ``` Call Singleton: ```scala val consumer = system.actorOf( ClusterSingletonProxy.props( singletonManagerPath = "/user/consumer", settings = ClusterSingletonProxySettings(system)), name = "consumerProxy") consumer ! SomeMessage ``` --- # Akka Clustering - Sharding Features: - One of the most powerful Akka features! - Allows to route messages across nodes in a cluster using a sharding function (actually two) - You don't need to know the physical location of an actor - cluster will forward message to a remote node if needed - Uses Akka Persistence internally (or brand-new Distributed Data) Concepts: - Coordinator - Shard Region - Entity Entities (actors) are "activated" by receiving a first message and can be "passivated" using `context.setReceiveTimeout`. ??? - Cluster has only one coordinator as a singleton - Ideal number of shards = max(nodes) * 10 - Every shard contains set of entities --- # Akka Clustering - Sharding Counter interface: ```scala case object Increment case object Decrement final case class Get(counterId: Long) final case class EntityEnvelope(id: Long, payload: Any) case object Stop final case class CounterChanged(delta: Int) ``` --- # Akka Clustering - Sharding Counter implementation: ```scala class Counter extends PersistentActor { context.setReceiveTimeout(120.seconds) override def persistenceId: String = "Counter-" + self.path.name var count = 0 def updateState(event: CounterChanged): Unit = count += event.delta override def receiveRecover: Receive = { case evt: CounterChanged ⇒ updateState(evt) } override def receiveCommand: Receive = { case Increment ⇒ persist(CounterChanged(+1))(updateState) case Decrement ⇒ persist(CounterChanged(-1))(updateState) case Get(_) ⇒ sender() ! count case ReceiveTimeout ⇒ context.parent ! Passivate(stopMessage = Stop) case Stop ⇒ context.stop(self) } } ``` --- # Akka Clustering - Sharding Create a region on every node: ```scala val counterRegion: ActorRef = ClusterSharding(system).start( typeName = "Counter", entityProps = Props[Counter], settings = ClusterShardingSettings(system), extractEntityId = extractEntityId, extractShardId = extractShardId) ``` Sharding functions: ```scala val extractEntityId: ShardRegion.ExtractEntityId = { case EntityEnvelope(id, payload) ⇒ (id.toString, payload) case msg @ Get(id) ⇒ (id.toString, msg) } val numberOfShards = 100 val extractShardId: ShardRegion.ExtractShardId = { case EntityEnvelope(id, _) ⇒ (id % numberOfShards).toString case Get(id) ⇒ (id % numberOfShards).toString } ``` --- class: center, middle # Akka Cluster Sharding + Persistence = ❤ --- class: center, middle # Testing in Akka --- # ScalaTest To include: > libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.1" % "test" Example test: ```scala import org.scalatest.{FlatSpec, Matchers} class Thing extends FlatSpec with Matchers { it should "work" in { Something.method("foo") should be ("bar") } } ``` Example run: ``` $ sbt test [info] Thing: [info] - should work [info] Run completed in 297 milliseconds. [info] Total number of tests run: 1 [info] Suites: completed 1, aborted 0 [info] Tests: succeeded 1, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. [success] Total time: 2 s, completed 2-Apr-2017 9:42:27 PM ``` --- # ScalaTest - Matchers http://www.scalatest.org/user_guide/using_matchers ```scala result should equal (3) // can customize equality result should === (3) // can customize equality and enforce type constraints result should be (3) // cannot customize equality, so fastest to compile result shouldEqual 3 // can customize equality, no parentheses required result shouldBe 3 // cannot customize equality, so fastest to compile, no parentheses required ``` ```scala result should have size 10 ``` ```scala string should endWith regex "wo.ld" ``` ```scala List(1, 2, 3, 4, 5) should contain oneOf (5, 7, 9) ``` ```scala case class Name(first: String, middle: String, last: String) val name = Name("Jane", "Q", "Programmer") inside(name) { case Name(first, _, _) => first should startWith ("S") } ``` ```scala an [IndexOutOfBoundsException] should be thrownBy s.charAt(-1) ``` --- # ScalaTest - Testing styles - FunSuite - **FlatSpec** - FunSpec - WordSpec - FreeSpec - PropSpec - FeatureSpec - RefSpec --- # ScalaTest - FunSuite ```scala import org.scalatest.FunSuite class SetSuite extends FunSuite { test("An empty Set should have size 0") { assert(Set.empty.size == 0) } test("Invoking head on an empty Set should produce NoSuchElementException") { assertThrows[NoSuchElementException] { Set.empty.head } } } ``` --- # ScalaTest - FlatSpec ```scala import org.scalatest.FlatSpec class SetSpec extends FlatSpec { "An empty Set" should "have size 0" in { assert(Set.empty.size == 0) } it should "produce NoSuchElementException when head is invoked" in { assertThrows[NoSuchElementException] { Set.empty.head } } } ``` --- # ScalaTest - FunSpec ```scala import org.scalatest.FunSpec class SetSpec extends FunSpec { describe("A Set") { describe("when empty") { it("should have size 0") { assert(Set.empty.size == 0) } it("should produce NoSuchElementException when head is invoked") { assertThrows[NoSuchElementException] { Set.empty.head } } } } } ``` --- # ScalaTest - Before and After ```scala import org.scalatest._ class ExampleSpec extends FlatSpec with BeforeAndAfter with BeforeAndAfterAll { before { // ... } after { // ... } override def beforeAll() { // ... } override def afterAll() { // ... } } ``` --- # ScalaTest - Advanced features - Fixtures - Mocks and stubs (using ScalaMock) - Property testing (using ScalaCheck) - Selenium DSL! - Async support (using `Eventually` helper) --- # Akka TestKit To include: > libraryDependencies += "com.typesafe.akka" %% "akka-testkit" % "2.4.17" % "test" Supports two types of tests: - "**Unit**" tests: synchronous, no concurrency, using classes - Not broadly used, have lots of limitations - Can give access to actor class methods and fields - "**Integration**" tests: asynchronous, using actors - Behave almost like in a real runtime (except Actor System is different) --- # Akka TestKit - Synchronous Testing Access internal state: ```scala import akka.testkit.TestActorRef val actorRef = TestActorRef(new MyActor) val actor = actorRef.underlyingActor actor.someField should be (123) ``` Synchronous execution: ```scala import akka.testkit.TestActorRef import scala.concurrent.duration._ import scala.concurrent.Await import akka.pattern.ask val actorRef = TestActorRef(new MyActor) // hypothetical message stimulating a '42' answer val future = actorRef ? Say42 val Success(result: Int) = future.value.get result should be (42) ``` --- # Akka TestKit - Asynchronous Testing ```scala class EchoActorSpec extends TestKit(ActorSystem("TestSystem")) with ImplicitSender with FlatSpecLike with Matchers { it should "send back messages unchanged" in { val echo = system.actorOf(TestActors.echoActorProps) echo ! "hello world" expectMsg("hello world") } } ``` ```scala val worker = system.actorOf(Props[Worker]) within(200 millis) { worker ! "some work" expectMsg("some result") expectNoMsg // will block for the rest of the 200ms Thread.sleep(300) // will NOT make this block fail } ``` --- # Akka TestKit - Asynchronous Testing `TestProbe` is great! ```scala val probe1 = TestProbe() val probe2 = TestProbe() val actor = system.actorOf(Props[MyDoubleEcho]) actor ! ((probe1.ref, probe2.ref)) actor ! "hello" probe1.expectMsg(500 millis, "hello") probe2.expectMsg(500 millis, "hello") ``` ```scala val probe = TestProbe() val future = probe.ref ? "hello" probe.expectMsg(0 millis, "hello") probe.reply("world") ``` --- # Akka TestKit - Asynchronous Testing Asserts: ```scala // expectMsg actor.expectMsg(Message("123")) // expectMsgPF actor.expectMsgPF() { case SomethingCreated(true, _, _, _, Some(account.clientId)) => true } // expectMsgClass actor.expectMsgClass(classOf[Message]) // receiveN actor.receiveN(100, 20 seconds) ``` --- # Summary - Scala is great! And I hope you think the same 😉. It can be very concise, but flexible and efficient. Especially if you follow FP principles - Actors and Futures are better concepts to use for concurrency - Akka is a great toolkit for building distributed systems - Try various things! Akka can look very complicated, but it's not that scary when you try it - Read papers ;-) --- # Links - https://www.scala-lang.org - http://blog.tmorris.net/posts/scalaoption-cheat-sheet/ - https://www.scala-exercises.org - https://www.lightbend.com/activator/download - http://stackoverflow.com/questions/29068064/scala-concurrent-blocking-what-does-it-actually-do - http://slick.lightbend.com - http://akka.io - https://github.com/sap1ens/akka-cluster-consul - http://camel.apache.org - http://www.reactive-streams.org - https://github.com/boldradius/akka-dddd-template - https://www.slideshare.net/sap1ens/building-eventing-systems-for-microservice-architecture - https://github.com/akka/alpakka - http://www.enterpriseintegrationpatterns.com - https://www.youtube.com/watch?v=H0i_bXKwujQ - http://www.scalatest.org --- class: center, middle # We did it! Thank you! Please answer quick servey: https://goo.gl/forms/QySYwiOLzwg4605p2


Running an Analysis¶
The first step in running a cross-section analysis is the creation of a
CrossSection
object. This class
stores the structural geometry and finite element mesh and provides methods to
perform various types of cross-section analyses.
-
class
sectionproperties.analysis.cross_section.
CrossSection
(geometry, mesh, materials=None, time_info=False)[source] Bases:
object
Class for structural cross-sections.
Stores the finite element geometry, mesh and material information and provides methods to compute the cross-section properties. The element type used in this program is the six-noded quadratic triangular element.
The constructor extracts information from the provided mesh object and creates and stores the corresponding Tri6 finite element objects.
Parameters: - geometry (
Geometry
) – Cross-section geometry object used to generate the mesh - mesh (
meshpy.triangle.MeshInfo
) – Mesh object returned by meshpy - materials (list[
Material
]) – A list of material properties corresponding to various regions in the geometry and mesh. Note that if materials are specified, the number of material objects must equal the number of regions in the geometry. If no materials are specified, only a purely geometric analysis can take place, and all regions will be assigned a default material with an elastic modulus and yield strength equal to 1, and a Poisson’s ratio equal to 0. - time_info (bool) – If set to True, a detailed description of the computation and the time cost is printed to the terminal.
The following example creates a
CrossSection
object of a 100D x 50W rectangle using a mesh size of 5:import sectionproperties.pre.sections as sections from sectionproperties.analysis.cross_section import CrossSection geometry = sections.RectangularSection(d=100, b=50) mesh = geometry.create_mesh(mesh_sizes=[5]) section = CrossSection(geometry, mesh)
The following example creates a 100D x 50W rectangle, with the top half of the section comprised of timber and the bottom half steel. The timber section is meshed with a maximum area of 10 and the steel section mesh with a maximum area of 5:
import sectionproperties.pre.sections as sections from sectionproperties.pre.pre import Material from sectionproperties.analysis.cross_section import CrossSection geom_steel = sections.RectangularSection(d=50, b=50) geom_timber = sections.RectangularSection(d=50, b=50, shift=[0, 50]) geometry = sections.MergedSection([geom_steel, geom_timber]) geometry.clean_geometry() mesh = geometry.create_mesh(mesh_sizes=[5, 10]) steel = Material(name='Steel', elastic_modulus=200e3, poissons_ratio=0.3, yield_strength=250, color='grey') timber = Material(name='Timber', elastic_modulus=8e3, poissons_ratio=0.35, yield_strength=20, color='burlywood') section = CrossSection(geometry, mesh, [steel, timber]) section.plot_mesh(materials=True, alpha=0.5)
Variables: - elements (list[
Tri6
]) – List of finite element objects describing the cross-section mesh - num_nodes (int) – Number of nodes in the finite element mesh
- geometry (
Geometry
) – Cross-section geometry object used to generate the mesh - mesh (
meshpy.triangle.MeshInfo
) – Mesh object returned by meshpy - mesh_nodes (
numpy.ndarray
) – Array of node coordinates from the mesh - mesh_elements (
numpy.ndarray
) – Array of connectivities from the mesh - mesh_attributes (
numpy.ndarray
) – Array of attributes from the mesh - materials – List of materials
- material_groups – List of objects containing the elements in each defined material
- section_props (
SectionProperties
) – Class to store calculated section properties
Raises: AssertionError – If the number of materials does not equal the number of regions
- geometry (
Checking the Mesh Quality¶
Before carrying out a cross-section analysis it is a good idea to check the quality of the finite element mesh. Some useful methods are provided to display mesh statistics and to plot the finite element mesh:
-
CrossSection.
display_mesh_info
()[source] Prints mesh statistics (number of nodes, elements and regions) to the command window.
The following example displays the mesh statistics for a Tee section merged from two rectangles:
import sectionproperties.pre.sections as sections from sectionproperties.analysis.cross_section import CrossSection rec1 = sections.RectangularSection(d=100, b=25, shift=[-12.5, 0]) rec2 = sections.RectangularSection(d=25, b=100, shift=[-50, 100]) geometry = sections.MergedSection([rec1, rec2]) mesh = geometry.create_mesh(mesh_sizes=[5, 2.5]) section = CrossSection(geometry, mesh) section.display_mesh_info() >>>Mesh Statistics: >>>--4920 nodes >>>--2365 elements >>>--2 regions
-
CrossSection.
plot_mesh
(ax=None, pause=True, alpha=1, materials=False, mask=None)[source] Plots the finite element mesh. If no axes object is supplied a new figure and axis is created.
Parameters: - ax (
matplotlib.axes.Axes
) – Axes object on which the mesh is plotted - pause (bool) – If set to true, the figure pauses the script until the window is closed. If set to false, the script continues immediately after the window is rendered.
- alpha (float) – Transparency of the mesh outlines: \(0 \leq \alpha \leq 1\)
- materials (bool) – If set to true and material properties have been
provided to the
CrossSection
object, shades the elements with the specified material colours - mask (list[bool]) – Mask array, of length
num_nodes
, to mask out triangles
The following example plots the mesh generated for the second example listed under the
CrossSection
object definition:import sectionproperties.pre.sections as sections from sectionproperties.pre.pre import Material from sectionproperties.analysis.cross_section import CrossSection geom_steel = sections.RectangularSection(d=50, b=50) geom_timber = sections.RectangularSection(d=50, b=50, shift=[50, 0]) geometry = sections.MergedSection([geom_steel, geom_timber]) geometry.clean_geometry() mesh = geometry.create_mesh(mesh_sizes=[5, 10]) steel = Material(name='Steel', elastic_modulus=200e3, poissons_ratio=0.3, yield_strength=250, color='grey') timber = Material(name='Timber', elastic_modulus=8e3, poissons_ratio=0.35, yield_strength=20, color='burlywood') section = CrossSection(geometry, mesh, [steel, timber]) section.plot_mesh(materials=True, alpha=0.5)
- ax (
Geometric Analysis¶
A geometric analysis calculates the area properties of the cross-section.
-
CrossSection.
calculate_geometric_properties
(time_info=False)[source] Calculates the geometric properties of the cross-section and stores them in the
SectionProperties
object contained in thesection_props
class variable.Parameters: time_info (bool) – If set to True, a detailed description of the computation and the time cost is printed to the terminal. The following geometric section properties are calculated:
- Cross-sectional area
- Modulus weighted area (axial rigidity)
- First moments of area
- Second moments of area about the global axis
- Second moments of area about the centroidal axis
- Elastic centroid
- Centroidal section moduli
- Radii of gyration
- Principal axis properties
If materials are specified for the cross-section, the moments of area and section moduli are elastic modulus weighted.
The following example demonstrates the use of this method:
section = CrossSection(geometry, mesh) section.calculate_geometric_properties()
Plastic Analysis¶
A plastic analysis calculates the plastic properties of the cross-section.
Note
A geometric analysis must be performed on the CrossSection object before a plastic analysis is carried out.
-
CrossSection.
calculate_plastic_properties
(pc_region=[0.95, 0.95, 0.95, 0.95], time_info=False, verbose=False, debug=False)[source] Calculates the plastic properties of the cross-section and stores the, in the
SectionProperties
object contained in thesection_props
class variable.Parameters: - pc_region (list[float, float, float, float]) – Region of cross-section in which to search for the plastic centroids. The value represents the fraction of the depth to search within perpendicular to the axis being considered [x-axis, y-axis, 11-axis, 22-axis]. Ensure 0 < pc_region[i] < 1. The default scheme searches 95% of the section depth (ie. centroid \(\pm\) 47.5%)
- time_info (bool) – If set to True, a detailed description of the computation and the time cost is printed to the terminal.
- verbose (bool) – If set to True, the number of iterations required for each plastic axis is printed to the terminal.
- debug (bool) – If set to True, the geometry is plotted each time a new mesh is generated by the plastic centroid algorithm.
The following warping section properties are calculated:
- Plastic centroid for bending about the centroidal and principal axes
- Plastic section moduli for bending about the centroidal and principal axes
- Shape factors for bending about the centroidal and principal axes
If materials are specified for the cross-section, the plastic section moduli are displayed as plastic moments (i.e \(M_p = f_y S\)) and the shape factors are not calculated.
Note that the geometric properties must be calculated before the plastic properties are calculated:
section = CrossSection(geometry, mesh) section.calculate_geometric_properties() section.calculate_plastic_properties()
Raises: RuntimeError – If the geometric properties have not been calculated prior to calling this method
Warping Analysis¶
A warping analysis calculates the torsion and shear properties of the cross-section.
Note
A geometric analysis must be performed on the CrossSection object before a warping analysis is carried out.
-
CrossSection.
calculate_warping_properties
(time_info=False, solver_type='direct')[source] Calculates all the warping properties of the cross-section and stores them in the
SectionProperties
object contained in thesection_props
class variable.Parameters: - time_info (bool) – If set to True, a detailed description of the computation and the time cost is printed to the terminal.
- solver_type (string) – Solver used for solving systems of linear equations, either using the ‘direct’ method or ‘cgs’ iterative method
The following warping section properties are calculated:
- Torsion constant
- Shear centre
- Shear area
- Warping constant
If materials are specified, the values calculated for the torsion constant, warping constant and shear area are elastic modulus weighted.
Note that the geometric properties must be calculated first for the calculation of the warping properties to be correct:
section = CrossSection(geometry, mesh) section.calculate_geometric_properties() section.calculate_warping_properties()
Raises: RuntimeError – If the geometric properties have not been calculated prior to calling this method
Stress Analysis¶
A stress analysis calculates the cross-section stresses arising from a set of forces
and moments. Executing this method returns a StressResult
object which stores the cross-section stresses and provides stress plotting functions.
Note
A geometric and warping analysis must be performed on the CrossSection object before a stress analysis is carried out.
-
CrossSection.
calculate_stress
(N=0, Vx=0, Vy=0, Mxx=0, Myy=0, M11=0, M22=0, Mzz=0, time_info=False)[source] Calculates the cross-section stress resulting from design actions and returns a
StressPost
object allowing post-processing of the stress results.Parameters: - N (float) – Axial force
- Vx (float) – Shear force acting in the x-direction
- Vy (float) – Shear force acting in the y-direction
- Mxx (float) – Bending moment about the centroidal xx-axis
- Myy (float) – Bending moment about the centroidal yy-axis
- M11 (float) – Bending moment about the centroidal 11-axis
- M22 (float) – Bending moment about the centroidal 22-axis
- Mzz (float) – Torsion moment about the centroidal zz-axis
- time_info (bool) – If set to True, a detailed description of the computation and the time cost is printed to the terminal.
Returns: Object for post-processing cross-section stresses
Return type: Note that a geometric and warping analysis must be performed before a stress analysis is carried out:
section = CrossSection(geometry, mesh) section.calculate_geometric_properties() section.calculate_warping_properties() stress_post = section.calculate_stress(N=1e3, Vy=3e3, Mxx=1e6)
Raises: RuntimeError – If a geometric and warping analysis have not been performed prior to calling this method
Calculating Frame Properties¶
Calculates the cross-section properties required for a 2D or 3D frame analysis.
Note
This method is significantly faster than performing a geometric and a warping analysis and has no prerequisites.
-
CrossSection.
calculate_frame_properties
(time_info=False, solver_type='direct')[source] Calculates and returns the properties required for a frame analysis. The properties are also stored in the
SectionProperties
object contained in thesection_props
class variable.Parameters: - time_info (bool) – If set to True, a detailed description of the computation and the time cost is printed to the terminal.
- solver_type (string) – Solver used for solving systems of linear equations, either using the ‘direct’ method or ‘cgs’ iterative method
Returns: Cross-section properties to be used for a frame analysis (area, ixx, iyy, ixy, j, phi)
Return type: tuple(float, float, float, float, float, float)
The following section properties are calculated:
- Cross-sectional area (area)
- Second moments of area about the centroidal axis (ixx, iyy, ixy)
- Torsion constant (j)
- Principal axis angle (phi)
If materials are specified for the cross-section, the area, second moments of area and torsion constant are elastic moulus weighted.
The following example demonstrates the use of this method:
section = CrossSection(geometry, mesh) (area, ixx, iyy, ixy, j, phi) = section.calculate_frame_properties()


public interface EventHandler
A user can choose to implement the EventHandler class, and implement handlers for all of the event types. Alternatively it is possible to extend the EventHandlerAdapter class, and implement only those events relevant to the specific application.
The EnvironmentConfig.setEventHandler
is used to provide
a mechanism for reporting event messages from the Berkeley DB library
to the application.
Berkeley DB is not re-entrant. Callback functions should not attempt to make library calls (for example, to release locks or close open handles). Re-entering Berkeley DB is not guaranteed to work correctly, and the results are undefined.
| Modifier and Type | Method and Description |
|---|---|
void |
handlePanicEvent()
A callback function to be called when a panic event is sent from the Berkeley DB library.
|
void |
handleRepAutoTakeoverFailedEvent()
A callback function to be called when a Replication automatic takeover
failed event is sent from the Berkeley DB library.
|
void |
handleRepClientEvent()
A callback function to be called when a Replication Client event is sent
from the Berkeley DB library.
|
void |
handleRepConnectBrokenEvent()
A callback function to be called when a Replication connect broken event
is sent from the Berkeley DB library.
|
void |
handleRepConnectEstablishedEvent()
A callback function to be called when a Replication connect established
event is sent from the Berkeley DB library.
|
void |
handleRepConnectTryFailedEvent()
A callback function to be called when a Replication connect retry failed
event is sent from the Berkeley DB library.
|
void |
handleRepDupmasterEvent()
A callback function to be called when a Replication Client event is sent
from the Berkeley DB library.
|
void |
handleRepElectedEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepElectionFailedEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepInitDoneEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepInQueueFullEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepJoinFailureEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepLocalSiteRemovedEvent()
A callback function to be called when a local site removed event is sent
from the Berkeley DB library.
|
void |
handleRepMasterEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepMasterFailureEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepNewMasterEvent(int envId)
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepPermFailedEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleRepSiteAddedEvent()
A callback function to be called when a site added event is sent from the
Berkeley DB library.
|
void |
handleRepSiteRemovedEvent()
A callback function to be called when a site removed event is sent from the
Berkeley DB library.
|
void |
handleRepStartupDoneEvent()
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void |
handleWriteFailedEvent(int errorCode)
A callback function to be called when an event is sent from the
Berkeley DB library.
|
void handlePanicEvent()
This event callback is received when an error occurs in the Berkeley DB
library where the only solution is to shut down the application and run
recovery. In such cases, the Berkeley DB methods will throw
RunRecoveryException
exceptions. It is often easier to simply exit
the application when such errors occur, rather than gracefully return up
the stack.
When this callback is received the database environment has failed. All threads of control in the database environment should exit the environment and recovery should be run.
void handleRepAutoTakeoverFailedEvent()
This event callback is received in a replication manager subordinate process when it fails to take over as the replication process and all replication manager threads in it are stopped unexpectedly.
void handleRepClientEvent()
This event callback is received when this member of a replication group is now a client site.
void handleRepConnectBrokenEvent()
This event callback is received when a previously established connection has been broken.
void handleRepConnectEstablishedEvent()
This event callback is received when a connection with a remote site has been established.
void handleRepConnectTryFailedEvent()
This event callback is received when an attempt to establish a new connection to a known remote site failed.
void handleRepDupmasterEvent()
This event callback is received when a duplicate master site has been discovered in the replication group.
void handleRepElectedEvent()
This event callback is received when this site has just won an election. An
Application using the Base replication API should arrange for a call to
the Environment.startReplication
method after receiving this
event to, reconfigure the local environment as a replication master.
Replication Manager applications may safely igore this event. The
Replication Manager calls Environment.startReplication
automatically on behalf of the application when appropriate (resulting in
firing of the EventHandler.handleRepMasterEvent
event).
void handleRepElectionFailedEvent()
This event callback is received when the local site's attempt to initiate or participate in a replication master election failed, due to the lack of timely message response from a sufficient number of remote sites.
void handleRepInitDoneEvent()
This event callback is received when internal initialization has been completed.
void handleRepInQueueFullEvent()
This event callback is received when replication Mananger incoming queue has reached its maximum threshold.
void handleRepJoinFailureEvent()
This event callback is received when the local site could not synchronize
with the master because an internal initialization was required, but
internal initialization has been turned off by the ReplicationConfig.AUTOINIT
flag to Environment.setReplicationConfig
void handleRepLocalSiteRemovedEvent()
This event callback is received when the local site has been removed from the group.
void handleRepMasterEvent()
This event callback is received when this site is now the master site of its replication group. It is the application's responsibility to begin acting as the master environment.
void handleRepMasterFailureEvent()
This event callback is received when the connection to the remote master replication site has failed.
void handleRepNewMasterEvent(int envId)
This event callback is received when the replication group of which this site is a member has just established a new master; the local site is not the new master.
envId
- The environment ID of the new master site.void handleRepPermFailedEvent()
This event callback is received when the replication manager did not
receive enough acknowledgements (based on the acknowledgement policy
configured with EnvironmentConfig.setReplicationManagerAckPolicy
)
to ensure a transaction's durability within the replication group. The
transaction will be flushed to the master's local disk storage for
durability.
void handleRepSiteAddedEvent()
This event callback is received when a new site has joined the group.
void handleRepSiteRemovedEvent()
This event callback is received when an existing remote site has been removed from the group.
void handleRepStartupDoneEvent()
This event callback is received when the client has completed startup synchronization and is now processing live log records received from the master.
void handleWriteFailedEvent(int errorCode)
This event callback is received when a Berkeley DB write to stable storage failed.
errorCode
- If an operating system specific error code is available for the failure it
will be passed in the errorCode parameter.Copyright (c) 1996, 2016 Oracle and/or its affiliates. All rights reserved.