If you read my previous story about Scalachain, you probably noticed that it is far from being a distributed system. It lacks all the features to properly work with other nodes. Add to it that a blockchain composed by a single node is useless. For this reason I decided it is time to work on the issue.

Since Scalachain is powered by Akka, why not take the chance to play with Akka Cluster? I created a simple project to tinker a bit with Akka Cluster, and in this story I’m going to share my learnings. We are going to create a cluster of three nodes, using Cluster Aware Routers to balance the load among them. Everything will run in a Docker container, and we will use docker-compose for an easy deployment.

Ok, Let’s roll! ?

Quick introduction to Akka Cluster

Akka Cluster provides great support to the creation of distributed applications. The best use case is when you have a node that you want to replicate N times in a distributed environment. This means that all the N nodes are peers running the same code. Akka Cluster gives you out-of-the-box the discovery of members in the same cluster. Using Cluster Aware Routers it is possible to balance the messages between actors in different nodes. It is also possible to choose the balancing policy, making load-balancing a piece of cake!

Actually you can chose between two types of routers:

Group Router — The actors to send the messages to — called routees — are specified using their actor path. The routers share the routees created in the cluster. We will use a Group Router in this example.

Group Router

Pool Router — The routees are created and deployed by the router, so they are its children in the actor hierarchy. Routees are not shared between routers. This is ideal for a primary-replica scenario, where each router is the primary and its routees the replicas.

Pool Router

This is just the tip of the iceberg, so I invite you to read the official documentation for more insights.

A Cluster for mathematical computations

Let’s picture a use-case scenario. Suppose to design a system to execute mathematical computations on request. The system is deployed online, so it needs a REST API to receive the computation requests. An internal processor handles these requests, executing the computation and returning the result.

Right now the processor can only compute the Fibonacci number. We decide to use a cluster of nodes to distribute the load among the nodes and improve performance. Akka Cluster will handle cluster dynamics and load-balancing between nodes. Ok, sounds good!

Actor hierarchy

First things first: we need to define our actor hierarchy. The system can be divided in three functional parts: the business logic, the cluster management, and the node itself. There is also the server but it is not an actor, and we will work on that later.

Business logic

The application should do mathematical computations. We can define a simple Processor actor to manage all the computational tasks. Every computation that we support can be implemented in a specific actor, that will be a child of the Processor one. In this way the application is modular and easier to extend and maintain. Right now the only child of Processor will be the ProcessorFibonacci actor. I suppose you can guess what its task is. This should be enough to start.

Cluster management

To manage the cluster we need a ClusterManager. Sounds simple, right? This actor handles everything related to the cluster, like returning its members when asked. It would be useful to log what happens inside the cluster, so we define a ClusterListener actor. This is a child of the ClusterManager, and subscribes to cluster events logging them.

Node

The Node actor is the root of our hierarchy. It is the entry point of our system that communicates with the API. The Processor and the ClusterManager are its children, along with the ProcessorRouter actor. This is the load balancer of the system, distributing the load among Processors. We will configure it as a Cluster Aware Router, so every ProcessorRouter can send messages to Processors on every node.

Actor hierarchy

Actor Implementation

Time to implement our actors! Fist we implement the actors related to the business logic of the system. We move then on the actors for the cluster management and the root actor (Node) in the end.

ProcessorFibonacci

This actor executes the computation of the Fibonacci number. It receives a Compute message containing the number to compute and the reference of the actor to reply to. The reference is important, since there can be different requesting actors. Remember that we are working in a distributed environment!

Once the Compute message is received, the fibonacci function computes the result. We wrap it in a ProcessorResponse object to provide information on the node that executed the computation. This will be useful later to see the round-robin policy in action.

The result is then sent to the actor we should reply to. Easy-peasy.

object ProcessorFibonacci {
  sealed trait ProcessorFibonacciMessage
  case class Compute(n: Int, replyTo: ActorRef) extends ProcessorFibonacciMessage

  def props(nodeId: String) = Props(new ProcessorFibonacci(nodeId))

  def fibonacci(x: Int): BigInt = {
    @tailrec def fibHelper(x: Int, prev: BigInt = 0, next: BigInt = 1): BigInt = x match {
      case 0 => prev
      case 1 => next
      case _ => fibHelper(x - 1, next, next + prev)
    }
    fibHelper(x)
  }
}

class ProcessorFibonacci(nodeId: String) extends Actor {
  import ProcessorFibonacci._

  override def receive: Receive = {
    case Compute(value, replyTo) => {
      replyTo ! ProcessorResponse(nodeId, fibonacci(value))
    }
  }
}

Processor

The Processor actor manages the specific sub-processors, like the Fibonacci one. It should instantiate the sub-processors and forward the requests to them. Right now we only have one sub-processor, so the Processor receives one kind of message: ComputeFibonacci. This message contains the Fibonacci number to compute. Once received, the number to compute is sent to a FibonacciProcessor, along with the reference of the sender().

object Processor {

  sealed trait ProcessorMessage

  case class ComputeFibonacci(n: Int) extends ProcessorMessage

  def props(nodeId: String) = Props(new Processor(nodeId))
}

class Processor(nodeId: String) extends Actor {
  import Processor._

  val fibonacciProcessor: ActorRef = context.actorOf(ProcessorFibonacci.props(nodeId), "fibonacci")

  override def receive: Receive = {
    case ComputeFibonacci(value) => {
      val replyTo = sender()
      fibonacciProcessor ! Compute(value, replyTo)
    }
  }
}

ClusterListener

We would like to log useful information about what happens in the cluster. This could help us to debug the system if we need to. This is the purpose of the ClusterListener actor. Before starting, it subscribes itself to the event messages of the cluster. The actor reacts to messages like MemberUp, UnreachableMember, or MemberRemoved, logging the corresponding event. When ClusterListener is stopped, it unsubscribes itself from the cluster events.

object ClusterListener {
  def props(nodeId: String, cluster: Cluster) = Props(new ClusterListener(nodeId, cluster))
}

class ClusterListener(nodeId: String, cluster: Cluster) extends Actor with ActorLogging {

  override def preStart(): Unit = {
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
      classOf[MemberEvent], classOf[UnreachableMember])
  }

  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case MemberUp(member) =>
      log.info("Node {} - Member is Up: {}", nodeId, member.address)
    case UnreachableMember(member) =>
      log.info(s"Node {} - Member detected as unreachable: {}", nodeId, member)
    case MemberRemoved(member, previousStatus) =>
      log.info(s"Node {} - Member is Removed: {} after {}",
        nodeId, member.address, previousStatus)
    case _: MemberEvent => // ignore
  }
}

ClusterManager

The actor responsible of the management of the cluster is ClusterManager. It creates the ClusterListener actor, and provides the list of cluster members upon request. It could be extended to add more functionalities, but right now this is enough.

object ClusterManager {

  sealed trait ClusterMessage
  case object GetMembers extends ClusterMessage

  def props(nodeId: String) = Props(new ClusterManager(nodeId))
}

class ClusterManager(nodeId: String) extends Actor with ActorLogging {

  val cluster: Cluster = Cluster(context.system)
  val listener: ActorRef = context.actorOf(ClusterListener.props(nodeId, cluster), "clusterListener")

  override def receive: Receive = {
    case GetMembers => {
      sender() ! cluster.state.members.filter(_.status == MemberStatus.up)
        .map(_.address.toString)
        .toList
    }
  }
}

ProcessorRouter

The load-balancing among processors is handled by the ProcessorRouter. It is created by the Node actor, but this time all the required information are provided in the configuration of the system.

class Node(nodeId: String) extends Actor {

  //...

  val processorRouter: ActorRef = context.actorOf(FromConfig.props(Props.empty), "processorRouter")

  //...
}

Let’s analyse the relevant part in the application.conf file.

akka {
  actor {
    ...
    deployment {
      /node/processorRouter {
        router = round-robin-group
        routees.paths = ["/user/node/processor"]
        cluster {
          enabled = on
          allow-local-routees = on
        }
      }
    }
  }
  ...
}

The first thing is to specify the path to the router actor, that is /node/processorRouter. Inside that property we can configure the behaviour of the router:

• router: this is the policy for the load balancing of messages. I chose the round-robin-group, but there are many others.
• routees.paths: these are the paths to the actors that will receive the messages handled by the router. We are saying: “When you receive a message, look for the actors corresponding to these paths. Choose one according to the policy and forward the message to it.” Since we are using Cluster Aware Routers, the routees can be on any node of the cluster.
• cluster.enabled: are we operating in a cluster? The answer is on, of course!
• cluster.allow-local-routees: here we are allowing the router to choose a routee in its node.

Using this configuration we can create a router to load balance the work among our processors.

Node

The root of our actor hierarchy is the Node. It creates the children actors — ClusterManager, Processor, and ProcessorRouter — and forwards the messages to the right one. Nothing complex here.

object Node {

  sealed trait NodeMessage

  case class GetFibonacci(n: Int)

  case object GetClusterMembers

  def props(nodeId: String) = Props(new Node(nodeId))
}

class Node(nodeId: String) extends Actor {

  val processor: ActorRef = context.actorOf(Processor.props(nodeId), "processor")
  val processorRouter: ActorRef = context.actorOf(FromConfig.props(Props.empty), "processorRouter")
  val clusterManager: ActorRef = context.actorOf(ClusterManager.props(nodeId), "clusterManager")

  override def receive: Receive = {
    case GetClusterMembers => clusterManager forward GetMembers
    case GetFibonacci(value) => processorRouter forward ComputeFibonacci(value)
  }
}

Server and API

Every node of our cluster runs a server able to receive requests. The Server creates our actor system and is configured through the application.conf file.

object Server extends App with NodeRoutes {

  implicit val system: ActorSystem = ActorSystem("cluster-playground")
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  val config: Config = ConfigFactory.load()
  val address = config.getString("http.ip")
  val port = config.getInt("http.port")
  val nodeId = config.getString("clustering.ip")

  val node: ActorRef = system.actorOf(Node.props(nodeId), "node")

  lazy val routes: Route = healthRoute ~ statusRoutes ~ processRoutes

  Http().bindAndHandle(routes, address, port)
  println(s"Node $nodeId is listening at http://$address:$port")

  Await.result(system.whenTerminated, Duration.Inf)

}

Akka HTTP powers the server itself and the REST API, exposing three simple endpoints. These endpoints are defined in the NodeRoutes trait.

The first one is /health, to check the health of a node. It responds with a 200 OK if the node is up and running

lazy val healthRoute: Route = pathPrefix("health") {
    concat(
      pathEnd {
        concat(
          get {
            complete(StatusCodes.OK)
          }
        )
      }
    )
  }

The /status/members endpoint responds with the current active members of the cluster.

lazy val statusRoutes: Route = pathPrefix("status") {
    concat(
      pathPrefix("members") {
        concat(
          pathEnd {
            concat(
              get {
                val membersFuture: Future[List[String]] = (node ? GetClusterMembers).mapTo[List[String]]
                onSuccess(membersFuture) { members =>
                  complete(StatusCodes.OK, members)
                }
              }
            )
          }
        )
      }
    )
  }

The last (but not the least) is the /process/fibonacci/n endpoint, used to request the Fibonacci number of n.

lazy val processRoutes: Route = pathPrefix("process") {
    concat(
      pathPrefix("fibonacci") {
        concat(
          path(IntNumber) { n =>
            pathEnd {
              concat(
                get {
                  val processFuture: Future[ProcessorResponse] = (node ? GetFibonacci(n)).mapTo[ProcessorResponse]
                  onSuccess(processFuture) { response =>
                    complete(StatusCodes.OK, response)
                  }
                }
              )
            }
          }
        )
      }
    )
  }

It responds with a ProcessorResponse containing the result, along with the id of the node where the computation took place.

Cluster Configuration

Once we have all our actors, we need to configure the system to run as a cluster! The application.conf file is where the magic takes place. I’m going to split it in pieces to present it better, but you can find the complete file here.

Let’s start defining some useful variables.

clustering {
  ip = "127.0.0.1"
  ip = ${?CLUSTER_IP}

  port = 2552
  port = ${?CLUSTER_PORT}

  seed-ip = "127.0.0.1"
  seed-ip = ${?CLUSTER_SEED_IP}

  seed-port = 2552
  seed-port = ${?CLUSTER_SEED_PORT}

  cluster.name = "cluster-playground"
}

Here we are simply defining the ip and port of the nodes and the seed, as well as the cluster name. We set a default value, then we override it if a new one is specified. The configuration of the cluster is the following.

akka {
  actor {
    provider = "cluster"
    ...
    /* router configuration */
    ...
  }
  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      hostname = ${clustering.ip}
      port = ${clustering.port}
    }
  }
  cluster {
    seed-nodes = [
      "akka.tcp://"${clustering.cluster.name}"@"${clustering.seed-ip}":"${clustering.seed-port}
    ]
    auto-down-unreachable-after = 10s
  }
}
...
/* server vars */
...
/* cluster vars */
}

Akka Cluster is build on top of Akka Remoting, so we need to configure it properly. First of all, we specify that we are going to use Akka Cluster saying that provider = "cluster". Then we bind cluster.ip and cluster.port to the hostname and port of the netty web framework.

The cluster requires some seed nodes as its entry points. We set them in the seed-nodes array, in the format akka.tcp://"{clustering.cluster.name}"@"{clustering.seed-ip}":”${clustering.seed-port}”. Right now we have one seed node, but we may add more later.

The auto-down-unreachable-after property sets a member as down after it is unreachable for a period of time. This should be used only during development, as explained in the official documentation.

Ok, the cluster is configured, we can move to the next step: Dockerization and deployment!

Dockerization and deployment

To create the Docker container of our node we can use sbt-native-packager. Its installation is easy: add addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.15") to the plugin.sbt file in the project/ folder. This amazing tool has a plugin for the creation of Docker containers. it allows us to configure the properties of our Dockerfile in the build.sbt file.

// other build.sbt properties

enablePlugins(JavaAppPackaging)
enablePlugins(DockerPlugin)
enablePlugins(AshScriptPlugin)

mainClass in Compile := Some("com.elleflorio.cluster.playground.Server")
dockerBaseImage := "java:8-jre-alpine"
version in Docker := "latest"
dockerExposedPorts := Seq(8000)
dockerRepository := Some("elleflorio")

Once we have setup the plugin, we can create the docker image running the command sbt docker:publishLocal. Run the command and taste the magic… ?

We have the Docker image of our node, now we need to deploy it and check that everything works fine. The easiest way is to create a docker-compose file that will spawn a seed and a couple of other nodes.

version: '3.5'

networks:
  cluster-network:

services:
  seed:
    networks:
      - cluster-network
    image: elleflorio/akka-cluster-playground
    ports:
      - '2552:2552'
      - '8000:8000'
    environment:
      SERVER_IP: 0.0.0.0
      CLUSTER_IP: seed
      CLUSTER_SEED_IP: seed

  node1:
    networks:
      - cluster-network
    image: elleflorio/akka-cluster-playground
    ports:
      - '8001:8000'
    environment:
      SERVER_IP: 0.0.0.0
      CLUSTER_IP: node1
      CLUSTER_PORT: 1600
      CLUSTER_SEED_IP: seed
      CLUSTER_SEED_PORT: 2552

  node2:
    networks:
      - cluster-network
    image: elleflorio/akka-cluster-playground
    ports:
      - '8002:8000'
    environment:
      SERVER_IP: 0.0.0.0
      CLUSTER_IP: node2
      CLUSTER_PORT: 1600
      CLUSTER_SEED_IP: seed
      CLUSTER_SEED_PORT: 2552

I won’t spend time going through it, since it is quite simple.

Let’s run it!

Time to test our work! Once we run the docker-compose up command, we will have a cluster of three nodes up and running. The seed will respond to requests at port :8000, while node1 and node2 at port :8001 and :8002. Play a bit with the various endpoints. You will see that the requests for a Fibonacci number will be computed by a different node each time, following a round-robin policy. That’s good, we are proud of our work and can get out for a beer to celebrate! ?

Conclusion

We are done here! We learned a lot of things in these ten minutes:

• What Akka Cluster is and what can do for us.
• How to create a distributed application with it.
• How to configure a Group Router for load-balancing in the cluster.
• How to Dockerize everything and deploy it using docker-compose.

You can find the complete application in my GitHub repo. Feel free to contribute or play with it as you like! ?

See you! ?

Akka HTTP’s routing DSL might seem complicated at first, but once you get the hang of it you’ll see how powerful it is.

In this tutorial we will focus on creating routes and their structure. We won’t cover parsing to and from JSON, we have other tutorials that cover that topic.

What are directives?

One of the first concepts we’ll find when learning server-side Akka HTTP (there’s a client-side library as well) is directives.

So, what are they?

You can think of them as building blocks, Lego pieces if you will, that you can use to construct your routes. They are composable, which means we can create directives on top of other directives.

If you want a more in-depth reading, feel free to check out Akka HTTP’s official documentation.

Before moving on, let’s discuss what we’ll build.

Blog-like API

We’ll create a sample of a public facing API for a blog, where we will allow users to:

• query a list of tutorials
• query a single tutorial by ID
• query the list of comments in a tutorial
• add comments to a tutorial

The endpoints will be:

- List all tutorials GET /tutorials

- Create a tutorial GET /tutorials/:id

- Get all comments in a tutorial GET /tutorials/:id/comments

- Add a comment to a tutorial POST /tutorials/:id/comments

We will only implement the endpoints, no logic in them. This way we’ll learn how to create this structure and the common pitfalls when starting with Akka HTTP.

Project Setup

We’ve created a repo for this tutorial, in it you’ll find a branch per each section that requires coding. Feel free to clone it and use it as a base project or even just change between branches to look at the differences.

Otherwise, create a new SBT project, and then add the dependencies in the build.sbt file:

name := "akkahttp-routing-dsl"

version := "0.1"

scalaVersion := "2.12.7"

val akkaVersion = "2.5.17" val akkaHttpVersion = "10.1.5"

libraryDependencies ++= Seq(   "com.typesafe.akka" %% "akka-actor" % akkaVersion,   "com.typesafe.akka" %% "akka-testkit" % akkaVersion % Test,  "com.typesafe.akka" %% "akka-stream" % akkaVersion,   "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % Test,   "com.typesafe.akka" %% "akka-http" % akkaHttpVersion,   "com.typesafe.akka" %% "akka-http-testkit" % akkaHttpVersion % Test,      "org.scalatest" %% "scalatest" % "3.0.5" % Test )

We added Akka HTTP and its dependencies, Akka Actor and Streams. And we will also use Scalatest for testing.

Listing all the tutorials

We’ll take a TDD approach to build our directive hierarchy, creating the tests first to make sure when don’t break our routes when adding others. Taking this approach is quite helpful when starting with Akka HTTP.

Let’s start with our route to listing all the tutorials. Create a new file under src/test/scala (if the folders don't exist, create them) named RouterSpec:

import akka.http.scaladsl.testkit.ScalatestRouteTest import org.scalatest.{Matchers, WordSpec}

class RouterSpec extends WordSpec with Matchers with ScalatestRouteTest {

}

WordSpec and Matchers are provided by Scalatest, and we'll use them to structure our tests and assertions. ScalatestRouteTest is a trait provided by Akka HTTP's test kit, it will allow us to test our routes in a convenient way. Let's see how we can accomplish that.

Because we’re using Scalatest’s WordSpec, we’ll start by creating a scope for our Router object that we will create soon and the first test:

"A Router" should {   "list all tutorials" in {   } }

Next, we want to make sure can send a GET request to the path /tutorials and get the response we expect, let's see how we can accomplish that:

Get("/tutorials") ~> Router.route ~> check {   status shouldBe StatusCodes.OK   responseAs[String] shouldBe "all tutorials" }

It won’t even compile because we haven’t created our Router object. Let's do that now.

Create a new Scala object under src/main/scala named Router. In it we will create a method that will return a Route:

import akka.http.scaladsl.server.Route

object Router {

  def route: Route = ???

}

Don’t worry too much about the ???, it's just a placeholder to avoid compilation errors temporarily. However, if that code is executed, it'll throw a NotImplementedError as we'll see soon.

Now that our tests and project are compiling, let’s run the tests (Right-click the spec and “Run ‘RouterSpec’”).

The test failed with the exception we were expecting, we haven’t implemented our routes. Let’s begin!

Creating the listing route

By looking into the official documentation we see that the route begins with the path directive. Let's mimic what they're doing and build our route:

import akka.http.scaladsl.server.{Directives, Route}

object Router extends Directives {

  def route: Route = path("tutorials") {    get {      complete("all tutorials")    }  }}

Seems reasonable, let’s run our spec. And it passes, great!

For reference, our entire RouterSpec now looks like:

import akka.http.scaladsl.model.StatusCodesimport akka.http.scaladsl.testkit.ScalatestRouteTestimport org.scalatest.{Matchers, WordSpec}class RouterSpec extends WordSpec with Matchers with ScalatestRouteTest {  "A Router" should {    "list all tutorials" in {      Get("/tutorials") ~> Router.route ~> check {        status shouldBe StatusCodes.OK        responseAs[String] shouldBe "all tutorials"      }    }  }}

Getting a single tutorial by ID

Next, we will allow our users to retrieve a single tutorial.

Let’s add a test for our new route:

"return a single tutorial by id" in {  Get("/tutorials/hello-world") ~> Router.route ~> check {    status shouldBe StatusCodes.OK    responseAs[String] shouldBe "tutorial hello-world"  }}

We expect to get back a message that includes the tutorial ID.

The test will fail because we haven’t created our route, let’s do that now.

From the same resource we used earlier to base our route on, we can see how we can place multiple directives at the same level in the hierarchy using the ~ directive.

We will have to nest path directives because need another segment after the /tutorials route for the tutorial ID. In the documentation they use IntNumber to extract a number from the path, but we'll use a string and for that we use can Segment instead.

Our route looks like:

def route: Route = path("tutorials") {  get {    complete("all tutorials")  } ~ path(Segment) { id =>    get {      complete(s"tutorial $id")    }  }}

Let’s run the tests. And you should get a similar error:

Request was rejectedScalaTestFailureLocation: RouterSpec at (RouterSpec.scala:17)org.scalatest.exceptions.TestFailedException: Request was rejected

What’s going on?!

Well, a request is rejected when it doesn’t match our directive hierarchy. This is one of the things that got me when starting.

Now is probably a good time to look into how these directives match the incoming request as it goes through the hierarchy.

Different directives will match different aspects of an incoming request, we’ve seen path and get, one matches the URL of the request and the other the method. If a request matches a directive it will go inside it, if it doesn't it will continue to the next one. This also tells us that order matters. If it doesn't match any directive the request is rejected.

Now that we now that our request is not matching our directives, let’s start looking into why.

If we look the documentation for the path directive (Cmd + Click on Mac) we'll find:

/** * Applies the given [[PathMatcher]] to the remaining unmatched path after consuming a leading slash. * The matcher has to match the remaining path completely. * If matched the value extracted by the [[PathMatcher]] is extracted on the directive level. * * @group path */

So, the path directive has to match exactly the path, meaning our first path directive will only match /tutorials and never /tutorials/:id.

In the same PathDirectives trait that contains the path directive we can see another directive named pathPrefix:

/** * Applies the given [[PathMatcher]] to a prefix of the remaining unmatched path after consuming a leading slash. * The matcher has to match a prefix of the remaining path. * If matched the value extracted by the PathMatcher is extracted on the directive level. * * @group path */

pathPrefix matches only a prefix and removes it. Sounds like this is what we're looking for, let's update our routes:

def route: Route = pathPrefix("tutorials") {  get {    complete("all tutorials")  } ~ path(Segment) { id =>    get {      complete(s"tutorial $id")    }  }}

Run the tests, and… we get another error. ?

"[all tutorials]" was not equal to "[tutorial hello-world]"ScalaTestFailureLocation: RouterSpec at (RouterSpec.scala:18)Expected :"[tutorial hello-world]"Actual   :"[all tutorials]"

Looks like our request matched the first get directive. It now matches the pathPrefix, and because it also is a GET request it will match the first get directive. Order matters.

There are a couple of things we can do. The simplest solution would be to move the first get request to the end of the hierarchy, however, we would have to remember this or document it. Not ideal.

Personally, I prefer avoiding such solutions and instead make the intend clear through code. If we look in the PathDirectives trait from earlier, we'll find a directive called pathEnd:

/** * Rejects the request if the unmatchedPath of the [[RequestContext]] is non-empty, * or said differently: only passes on the request to its inner route if the request path * has been matched completely. * * @group path */

That’s exactly what we want, so let’s wrap our first get directive with pathEnd:

def route: Route = pathPrefix("tutorials") {  pathEnd {    get {      complete("all tutorials")    }  } ~ path(Segment) { id =>    get {      complete(s"tutorial $id")    }  }}

Run the tests again, and… finally, the tests are passing! ?

Listing all comments in a tutorial

Let’s put into practice what we learned about nesting routes by taking it a bit further.

First the test:

"list all comments of a given tutorial" in {  Get("/tutorials/hello-world/comments") ~> Router.route ~> check {    status shouldBe StatusCodes.OK    responseAs[String] shouldBe "comments for the hello-world tutorial"  }}

It’s a similar case as before: we know we’ll need to place a route next to another one, which means we need to:

• change the path(Segmenter) to pathPrefix(Segmenter)
• wrap the first get with the pathEnd directive
• place the new route next to the pathEnd

Our routes end up looking like:

def route: Route = pathPrefix("tutorials") {  pathEnd {    get {      complete("all tutorials")    }  } ~ pathPrefix(Segment) { id =>    pathEnd {      get {        complete(s"tutorial $id")      }    } ~ path("comments") {      get {        complete(s"comments for the $id tutorial")      }    }  }}

Run the tests, and they should pass! ?

Adding comments to a tutorial

Our last endpoint is similar to the previous, but it will match POST requests. We’ll use this example to see the difference between implementing and testing a GET request versus a POST request.

The test:

"add comments to a tutorial" in {  Post("/tutorials/hello-world/comments", "new comment") ~> Router.route ~> check {    status shouldBe StatusCodes.OK    responseAs[String] shouldBe "added the comment 'new comment' to the hello-world tutorial"  }}

We’re using the Post method instead of the Get we've been using, and we're giving it an additional parameter which is the request body. The rest is familiar to us now.

To implement our last route, we can refer to the documentation and look at how it’s usually done.

We have a post directive just as we have a get one. To extract the request body we need two directives, entity and as, to which we supply the type we expect. In our case it's a string.

Let’s give that a try:

post {  entity(as[String]) { comment =>    complete(s"added the comment '$comment' to the $id tutorial")  }}

Looks reasonable. We extract the request body as a string and use it in our response. Let’s add it to our route method next to the previous route we worked on:

def route: Route = pathPrefix("tutorials") {  pathEnd {    get {      complete("all tutorials")    }  } ~ pathPrefix(Segment) { id =>    pathEnd {      get {        complete(s"tutorial $id")      }    } ~ path("comments") {      get {        complete(s"comments for the $id tutorial")      } ~ post {        entity(as[String]) { comment =>          complete(s"added the comment '$comment' to the $id tutorial")        }      }    }  }}

If you’d like to learn how to parse Scala classes to and from JSON we’ve got tutorials for that as well.

Run the tests, and they should all pass.

Conclusion

Akka HTTP’s routing DSL might seem confusing at first, but after overcoming some bumps it just clicks. After a while it’ll come naturally and it can be very powerful.

We learned how to structure our routes, but more importantly, we learned how to create that structure guided by tests which will make sure we don’t break them at some point in the future.

Even though we only worked on four endpoints, we ended up with a somewhat complicated and deep structure. Stay tuned and we’ll explore different ways to simplify our routes and make them more manageable!

Learn how to build REST APIs with Scala and Akka HTTP with this step-by-step free course!

Originally published at www.codemunity.io.

Synchronized is Java’s traditional concurrency mechanism. Although it is probably not something we see often these days, it is still fueling many libraries. The problem is, synchronized is both blocking and complicated. In this article, I’d like to illustrate the issue in a simple way and my reasoning to move to Akka Actor for better and easier concurrency.

Consider this simple code:

  int x; 

 if (x > 0) {
   return true;
 } else {
   return false;
 }

So return true if x is positive. Simple.

Next, consider this even simpler code:

x++;

Yes, a counter. Very simple right?

However, all these codes can blow up spectacularly in a multi-threaded environment.

In the first example, true or false isn’t determined by the value of x. It is actually determined by the if-test. So, if another thread changes x to a negative right after the first thread passed the if-test, we’d still get true even if x is no longer positive.

The second example is pretty deceptive. Although it is just one line, there are actually three operations: reading x, incrementing it and putting the updated value back. If two threads run at exactly the same time, the update might be lost.

When we have different threads simultaneously accessing and modifying a variable, we have a race condition. If we just want to build a counter, Java provides thread-safe Atomic Variables, among them Atomic Integer which we can use for this purpose. Atomic Integer, however, only works on single variables. How do we make several operations atomic?

By using synchronized block. First, let’s take a look at a more elaborate example.

int x; 
public int withdraw(int deduct){
    int balance = x - deduct; 
    if (balance > 0) {
      x = balance;
      return deduct;
    } else {
      return 0;
    }
}

This is a very basic cash withdraw method. It also happens to be dangerous. Two threads running at the same time may cause the bank to issue two withdrawals even if the balance is no longer enough. Now let’s see how it works with synchronized block:

volatile int x;
public int withdraw(int deduct){
  synchronized(this){
    int balance = x - deduct; 
    if (balance > 0) {
      x = balance;
      return deduct;
    } else {
      return 0;
    }
  }
}

The idea of synchronized block is simple. One thread enters it and locks it, while other threads wait outside. The lock is an object, in our case this. After it’s done, the lock is released and passed to another thread which then does the same thing. Also, note the esoteric keyword volatile which is needed to prevent the thread from using the local CPU cache of x.

Now with the threads untangled, the bank won’t accidentally issue unfunded withdrawals. However, this structure tends to grow complex with more blocks and more locks. Dealing with multiple locks is particularly risky. The blocks might inadvertently hold the key for each other and end up locking the entire app. And on top of it, we have an efficiency issue. Remember that while a thread works inside, all the other threads wait outside. And waiting threads are well … waiting. They don’t do anything else but wait.

So instead of doing such mechanism, why not just drop the job in a queue? To better visualize it, imagine an email system. When you send an email, you drop the email in the recipient’s mailbox. You don’t wait around until the person reads it.

These are the basics of the Actor model and Akka framework in general.

Actor encapsulates state and behavior. Unlike OOP’s encapsulation, though, actors do not expose their state and behavior at all. The only way for an actor to communicate with each other is by exchanging messages. Incoming messages are dropped in a mailbox and digested in first-in-first-out order. Here’s a reworked sample in Akka and Scala.

case class Withdraw(deduct: Int)
class ReplicaActor extends Actor {
  var x = 10;
  def receive: Receive = {
    case Withdraw(deduct) => val r = withdraw(deduct)
  }
}
class BossActor extends Actor {
  var replica = context.actorOf(Props[ReplicaActor])
  replica ! Withdraw(6)
  replica ! Withdraw(9)   
}

We have a ReplicaActor that does the work and BossActor that orders the replica around. First, notice the ! sign or tell. This is one of two methods (the other is ask) for an actor to asynchronously send a message to another actor. tell in particular does so without waiting for a reply. So the boss tells the replica to do two withdraw orders and immediately leaves. These messages arrive in the replica's receive where each one is popped and matched with the corresponding handler. In this case, Withdraw executes the withdraw method from the previous example and deducts the requested amount from state x. After it is done, the actor proceeds to the next message in the queue.

So what do we get here? For one, we no longer need to worry about locking and working with atomic/concurrent types. Actor’s encapsulation and queuing mechanism already guarantee thread-safety. And there is no more waiting since threads just drop the message and return. Results can be delivered later with ask or tell. It’s simple and sane.

Akka is based on JVM and available in both Scala and Java. Although this article isn’t debating Java vs Scala, Scala’s pattern matching and functional programming would be very useful in managing Actor’s data messaging. At the very least it can help you write shorter code by avoiding Java’s brackets and semicolons.