A Scala Project with Akka, Cats, and Cassandra

Data Modeling

This bank account actor will need a few pieces of data:

1. The messages it can receive — in Akka Persistence-speak they’re called commands
2. The events it will store in Cassandra
3. The structures to manage its internal state
4. The responses it may send to the main bank actor

The commands will look as follows:

sealed trait Command
object Command {
  case class CreateBankAccount(user: String, currency: String, initialBalance: Double, replyTo: ActorRef[Response]) extends Command
  case class UpdateBalance(id: String, currency: String, amount: Double /* can be < 0*/, replyTo: ActorRef[Response]) extends Command
  case class GetBankAccount(id: String, replyTo: ActorRef[Response]) extends Command
}

Obviously a simplification to how a production application would look like, but for the scope of this project should be complete enough. Notice all the commands have a replyTo: ActorRef[Response] in them, so that this actor knows who to send the response back to.

In terms of events, we need to store just enough so that we can reestablish the state of the bank account from these events if the application crashed for whatever reason. The data structures we’ll use will look like this:

trait Event
case class BankAccountCreated(bankAccount: BankAccount) extends Event
case class BalanceUpdated(amount: Double) extends Event

As for state, we’ll simply store the identifier of the account, the user’s identifier, the currency and the current amount.

case class BankAccount(id: String, user: String, currency: String, balance: Double)

In terms of responses, we’ll match them with the appropriate commands. They will look like this:

sealed trait Response
object Response {
  case class BankAccountCreatedResponse(id: String) extends Response
  case class BankAccountBalanceUpdatedResponse(maybeBankAccount: Try[BankAccount]) extends Response
  case class GetBankAccountResponse(maybeBankAccount: Option[BankAccount]) extends Response
}

Notice that in the BankAccountBalanceUpdatedResponse class we use a Try, because the updating might fail for different reasons, for example:

• The id of the bank account requested might be different from this bank account’s id
• The amount might be illegal, e.g. trying to withdraw more than you have

An in the GetBankAccountResponse we use an Option because there are only two answers we’re considering: either there is a bank account here, or it’s not.

The Bank Account Persistent Actor

A persistent actor is defined in terms of four things:

1. Its unique persistence ID, which will be used to store data in the store and retrieve data from the store
2. Its state, which can change over time
3. Its message handler, aka command handler
4. Its event handler, i.e. what the actor does after storing an event to the persistent store, or restoring an event after failure

Let’s take them in turn. First: the persistence ID — this one is straightforward, because the main bank actor will allocate a new UUID once it creates this actor in the first place, so we’ll simply pass it on upon creation.

Second: the state. We already know which data type we’ll use (BankAccount), so we’ll simply need to pass an “empty” state upon creation. Modifying the state happens when the actor receives a message (= a command) and when the actor handles an event, which are items 3 and 4 on the above list.

Third: the command handler. This is a function that, given the current state and an incoming command, will produce an Effect. This Effect may involve sending messages, persisting items in the persistent store, changing state, or a combination of the above. The function signature is as follows:

val commandHandler: (BankAccount, Command) => Effect[Event, BankAccount] = (state, command) => // continue here

and we can also continue with the implementation — we can run a pattern match on the command and treat each possible message type in turn:

// continued
command match {
  case CreateBankAccount(user, currency, initialBalance, bank) =>
    val id = state.id
    Effect
      .persist(BankAccountCreated(BankAccount(id, user, currency, initialBalance))) // persisted into Cassandra
      .thenReply(bank)(_ => BankAccountCreatedResponse(id))
  // continue here
}

There’s a lot of magic happening in these few lines. Breaking this down:

• Before this account receives any messages, it needs to be created by the bank.
• Upon creation of this account, the bank will send it a CreateBankAccount message.
• Upon reception of the command from the main bank actor, the account will store a BankAccountCreated event to Cassandra.
• After storing the event, the event handler will be invoked (the fourth item on the list, to be written shortly).
• The account will then reply to the bank actor with a BankAccountCreatedResponse.
• The bank will then surface the response to the HTTP layer, but that’s none of our concern right now.

Going forward with the other cases:

// continued
case UpdateBalance(_, _, amount, bank) =>
  val newBalance = state.balance + amount
  // check here for withdrawal
  if (newBalance < 0) // illegal
    Effect.reply(bank)(BankAccountBalanceUpdatedResponse(Failure(new RuntimeException("Cannot withdraw more than available"))))
  else
    Effect
      .persist(BalanceUpdated(amount))
      .thenReply(bank)(newState => BankAccountBalanceUpdatedResponse(Success(newState)))
case GetBankAccount(_, bank) =>
  Effect.reply(bank)(GetBankAccountResponse(Some(state)))
} // closing the pattern match

Following the structure of the first case,

• If we attempt withdrawing more than we have available, we’ll send back a response to the bank with a Failure
• If the balance modification was successful, persist the appropriate event and send back a successful response
• The “get” command simply responds with the current state of the account; can be improved with security checks, etc.

That was the command handler, the third item on the list.

Fourth: the event handler. This is a function which, given the current state of the actor and the event which has just been stored/restored, will return a new state of the actor. This assumes the event was successfully stored:

val eventHandler: (BankAccount, Event) => BankAccount = (state, event) =>
  event match {
    case BankAccountCreated(bankAccount) =>
      bankAccount
    case BalanceUpdated(amount) =>
      state.copy(balance = state.balance + amount)
  }

The final bit ties everything together:

def apply(id: String): Behavior[Command] =
  EventSourcedBehavior[Command, Event, BankAccount](
    persistenceId = PersistenceId.ofUniqueId(id),
    emptyState = BankAccount(id, "", "", 0.0), // unused
    commandHandler = commandHandler,
    eventHandler = eventHandler
  )

The complete code will look like this:

package com.rockthejvm.bank.actors

import akka.actor.typed.{ActorRef, Behavior}
import akka.persistence.typed.PersistenceId
import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior}

import scala.util.{Failure, Success, Try}

// a single bank account
object PersistentBankAccount {

  // commands = messages
  sealed trait Command
  object Command {
    case class CreateBankAccount(user: String, currency: String, initialBalance: Double, replyTo: ActorRef[Response]) extends Command
    case class UpdateBalance(id: String, currency: String, amount: Double /* can be < 0*/, replyTo: ActorRef[Response]) extends Command
    case class GetBankAccount(id: String, replyTo: ActorRef[Response]) extends Command
  }

  // events = to persist to Cassandra
  trait Event
  case class BankAccountCreated(bankAccount: BankAccount) extends Event
  case class BalanceUpdated(amount: Double) extends Event

  // state
  case class BankAccount(id: String, user: String, currency: String, balance: Double)

  // responses
  sealed trait Response
  object Response {
    case class BankAccountCreatedResponse(id: String) extends Response
    case class BankAccountBalanceUpdatedResponse(maybeBankAccount: Try[BankAccount]) extends Response
    case class GetBankAccountResponse(maybeBankAccount: Option[BankAccount]) extends Response
  }

  import Command._
  import Response._


  val commandHandler: (BankAccount, Command) => Effect[Event, BankAccount] = (state, command) =>
    command match {
      case CreateBankAccount(user, currency, initialBalance, bank) =>
        val id = state.id
        Effect
          .persist(BankAccountCreated(BankAccount(id, user, currency, initialBalance))) // persisted into Cassandra
          .thenReply(bank)(_ => BankAccountCreatedResponse(id))
      case UpdateBalance(_, _, amount, bank) =>
        val newBalance = state.balance + amount
        // check here for withdrawal
        if (newBalance < 0) // illegal
          Effect.reply(bank)(BankAccountBalanceUpdatedResponse(Failure(new RuntimeException("Cannot withdraw more than available"))))
        else
          Effect
            .persist(BalanceUpdated(amount))
            .thenReply(bank)(newState => BankAccountBalanceUpdatedResponse(Success(newState)))
      case GetBankAccount(_, bank) =>
        Effect.reply(bank)(GetBankAccountResponse(Some(state)))
    }

  val eventHandler: (BankAccount, Event) => BankAccount = (state, event) =>
    event match {
      case BankAccountCreated(bankAccount) =>
        bankAccount
      case BalanceUpdated(amount) =>
        state.copy(balance = state.balance + amount)
    }

  def apply(id: String): Behavior[Command] =
    EventSourcedBehavior[Command, Event, BankAccount](
      persistenceId = PersistenceId.ofUniqueId(id),
      emptyState = BankAccount(id, "", "", 0.0), // unused
      commandHandler = commandHandler,
      eventHandler = eventHandler
    )
}

Testing the Actors

For interaction with Cassandra, we’re going to need a configuration to use the Cassandra journal, so in src/main/resources, we’ll add an application.conf file with the following configuration:

# Journal
akka.persistence.journal.plugin = "akka.persistence.cassandra.journal"
akka.persistence.cassandra.journal.keyspace-autocreate = true
akka.persistence.cassandra.journal.tables-autocreate = true
datastax-java-driver.advanced.reconnect-on-init = true

# Snapshot
akka.persistence.snapshot-store.plugin = "akka.persistence.cassandra.snapshot"
akka.persistence.cassandra.snapshot.keyspace-autocreate = true
akka.persistence.cassandra.snapshot.tables-autocreate = true

akka.actor.allow-java-serialization = on

We can of course use the Akka TestKit to test the actors, but we’ll go for some live testing with some events stored in Cassandra! With the Docker container up and running — all you need is to run docker-compose up in the root directory — we’ll write some quick application to run with the Bank actor and some creation/retrieval commands:

object BankPlayground {
  import PersistentBankAccount.Command._
  import PersistentBankAccount.Response._
  import PersistentBankAccount.Response

  def main(args: Array[String]): Unit = {
    val rootBehavior: Behavior[NotUsed] = Behaviors.setup { context =>
      val bank = context.spawn(Bank(), "bank")
      val logger = context.log

      val responseHandler = context.spawn(Behaviors.receiveMessage[Response]{
        case BankAccountCreatedResponse(id) =>
          logger.info(s"successfully created bank account $id")
          Behaviors.same
        case GetBankAccountResponse(maybeBankAccount) =>
          logger.info(s"Account details: $maybeBankAccount")
          Behaviors.same
      }, "replyHandler")

      // ask pattern
      import akka.actor.typed.scaladsl.AskPattern._
      import scala.concurrent.duration._
      implicit val timeout: Timeout = Timeout(2.seconds)
      implicit val scheduler: Scheduler = context.system.scheduler
      implicit val ec: ExecutionContext = context.executionContext

      // test 1
      //      bank ! CreateBankAccount("daniel", "USD", 10, responseHandler)

      // test 2
      //      bank ! GetBankAccount("replaceWithYourUuidHere", responseHandler)

      Behaviors.empty
    }

    val system = ActorSystem(rootBehavior, "BankDemo")
  }
}

For this live test, we’re first going to send a creation message and check that there are two events in Cassandra (one from the bank and one from the account). Running this application should give us the log successfully created bank account .... You can then shut down the application and run it again, this time with just the second message. A successful log with the account details proves multiple things:

• That the bank actor works
• That the account actor works
• That the bank account can successfully respawn the account
• That the account can successfully restore its state

Sure enough, we can also inspect Cassandra for the relevant events. While Cassandra is running, from another terminal we can run

docker ps

and look at the container name, copy it, and then run

docker exec -it akka-cassandra-demo_cassandra_1 cqlsh

which will open the CQL prompt for us to inspect the tables. Inside, we’ll run

select * from akka.messages;

and lo and behold, we have messages there! The Cassandra tables were created automatically by the Akka Persistence Cassandra journal.

 persistence_id                       | partition_nr | sequence_nr | timestamp                            | event                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  | event_manifest | meta | meta_ser_id | meta_ser_manifest | ser_id | ser_manifest | tags | timebucket    | writer_uuid
--------------------------------------+--------------+-------------+--------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------------+------+-------------+-------------------+--------+--------------+------+---------------+--------------------------------------
                                 bank |            0 |           1 | 34633920-b1ce-11ec-8405-a1782b503565 |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     0xaced000573720032636f6d2e726f636b7468656a766d2e62616e6b2e6163746f72732e42616e6b2442616e6b4163636f756e74437265617465646aebebd03a53f5500200014c000269647400124c6a6176612f6c616e672f537472696e673b787074002434373166373833392d633363392d346532352d393035622d326333626533343132383437 |                | null |        null |              null |      1 |              | null | 1648825200000 | 0dd3a479-9563-488b-8b72-d259bbaf5f8f
 471f7839-c3c9-4e25-905b-2c3be3412847 |            0 |           1 | 346c87f0-b1ce-11ec-8405-a1782b503565 | 0xaced000573720043636f6d2e726f636b7468656a766d2e62616e6b2e6163746f72732e50657273697374656e7442616e6b4163636f756e742442616e6b4163636f756e744372656174656493750afb52eb6b5b0200014c000b62616e6b4163636f756e7474003e4c636f6d2f726f636b7468656a766d2f62616e6b2f6163746f72732f50657273697374656e7442616e6b4163636f756e742442616e6b4163636f756e743b78707372003c636f6d2e726f636b7468656a766d2e62616e6b2e6163746f72732e50657273697374656e7442616e6b4163636f756e742442616e6b4163636f756e74d653249c8fb35b6d02000444000762616c616e63654c000863757272656e63797400124c6a6176612f6c616e672f537472696e673b4c0002696471007e00044c00047573657271007e00047870402400000000000074000355534474002434373166373833392d633363392d346532352d393035622d32633362653334313238343774000664616e69656c |                | null |        null |              null |      1 |              | null | 1648825200000 | 4f9cea0e-13d3-403f-8574-15f19a3a5664

Let’s move on to the HTTP server.

Creating a Bank Account in the Akka HTTP Server

Starting with the first endpoint, a POST on /bank:

object BankRouter {
  val routes =
    pathPrefix("bank") {
      pathEndOrSingleSlash {
        post {
          // parse the payload
          entity(as[BankAccountCreationRequest]) { request =>
            /*
              TODO 1
             */
          }
        }
      }
    }
}

Inside, we need to do the following:

1. Fetch the bank actor
2. Send it a CreateBankAccount command¹
3. Parse its reply
4. Send back an HTTP response

First, we need the bank actor, which we don’t have. We can receive it as a constructor argument to this BankRouter, which means we’ll need to make it a class. Besides, we’ll also need an ActorSystem to be able to run the directives, so we’ll pass this one too, as an implicit argument, or a using clause in Scala 3.

// at the top
import akka.actor.typed.{ActorRef, ActorSystem}
import com.rockthejvm.bank.actors.PersistentBankAccount.Command
import com.rockthejvm.bank.actors.PersistentBankAccount.Command._

// change here
class BankRouter(bank: ActorRef[Command])(implicit system: ActorSystem[_]) {
  // same routes
}

For point number two, we need to convert the HTTP request into a command we can pass to the bank actor:

case class BankAccountCreationRequest(user: String, currency: String, balance: Double) {
  // added now
  def toCommand(replyTo: ActorRef[Response]): Command = CreateBankAccount(user, currency, balance, replyTo)
}

For point number three, we can use the bank actor to send a command and expect a reply. We’ll use the ask pattern to do this.

// at the top
import akka.actor.typed.scaladsl.AskPattern._
import akka.util.Timeout
import scala.concurrent.duration._

// within BankRouter
implicit val timeout: Timeout = Timeout(5.seconds)
def createBankAccount(request: BankAccountCreationRequest): Future[Response] =
  bank.ask(replyTo => request.toCommand(replyTo))

The ask pattern is useful for this kind of one-off, request-response interaction. Akka will create an intermediate actor with a short lifespan, which will serve as the destination for the eventual response, and this actor will fulfill a Future with that response when it receives it. It is this Future that we can then handle in our “regular”, non-actor code.

Finally, point four is our TODO 1:

// instead of TODO 1
onSuccess(createBankAccount(request)) {
  // send back an HTTP response
  case BankAccountCreatedResponse(id) =>
    respondWithHeader(Location(s"/bank/$id")) {
      complete(StatusCodes.Created)
    }
}

onSuccess is a directive that asynchronously waits for a Future to be completed, and once it’s done, the content of the Future is subject to the function below, which needs to return another directive: in our case, we’ll return an HTTP 201, and will return the URI of the bank account as a Location HTTP header.

Retrieving a Bank Account

Our routes currently look like this:

  val routes =
    pathPrefix("bank") {
      pathEndOrSingleSlash {
        post {
          // parse the payload
          entity(as[BankAccountCreationRequest]) { request =>
            onSuccess(createBankAccount(request)) {
              // send back an HTTP response
              case BankAccountCreatedResponse(id) =>
                respondWithHeader(Location(s"/bank/$id")) {
                  complete(StatusCodes.Created)
                }
            }
          }
        }
      }
    }

The second endpoint is a GET on /bank/someUUID, so we need to add another route inside pathPrefix("bank"):

  val routes =
    pathPrefix("bank") {
      pathEndOrSingleSlash {
        // same code
      } ~ // <-- careful with this one
        path(Segment) { id =>
          get {
            // TODO 2
          }
        }
    }

We need to make Akka HTTP parse the next token after /bank and return that token to us as the identifier of the account. Once again, we need to

1. Send a command to the bank actor to retrieve the details
2. Parse the response
3. Send back an HTTP response

We’ll follow the same pattern as before, so we’ll add a method to ask the bank actor for some account details:

  def getBankAccount(id: String): Future[Response] =
    bank.ask(replyTo => GetBankAccount(id, replyTo))

And we’ll parse the response and send back a proper HTTP response in TODO 2:

onSuccess(getBankAccount(id)) {
  case GetBankAccountResponse(Some(account)) =>
    complete(account)
  case GetBankAccountResponse(None) =>
    complete(StatusCodes.NotFound, FailureResponse(s"Bank account $id cannot be found."))
}

We complete the response with the account details passed directly as the case class instance, because the implicit marshallers will take care to serialize that instance to JSON.

Updating a Bank Account

Our routes now look like this:

  val routes =
    pathPrefix("bank") {
      pathEndOrSingleSlash {
        post {
          // parse the payload
          entity(as[BankAccountCreationRequest]) { request =>
            onSuccess(createBankAccount(request)) {
              // send back an HTTP response
              case BankAccountCreatedResponse(id) =>
                respondWithHeader(Location(s"/bank/$id")) {
                  complete(StatusCodes.Created)
                }
            }
          }
        }
      } ~
      path(Segment) { id =>
        get {
          onSuccess(getBankAccount(id)) {
            case GetBankAccountResponse(Some(account)) =>
              complete(account)
            case GetBankAccountResponse(None) =>
              complete(StatusCodes.NotFound, FailureResponse(s"Bank account $id cannot be found."))
          }
        }
      }
    }

We need to add a third endpoint, which is a PUT on the same /bank/UUID path, therefore:

  val routes =
    pathPrefix("bank") {
      pathEndOrSingleSlash {
        // endpoint 1
      } ~
      path(Segment) { id =>
        get {
          // endpoint 2
        } ~ // <-- watch this one
        put {
          entity(as[BankAccountUpdateRequest]) { request => // need to parse the request
            // TODO 3
          }
        }
      }
    }

Following the same pattern, in this endpoint we need to both parse the HTTP request’s payload and send back an HTTP response with a payload. So, same deal:

1. Ask the bank actor to update the bank account.
2. Expect a reply.
3. Send back an HTTP response.

Adding a method to ask the bank actor for the update:

def updateBankAccount(id: String, request: BankAccountUpdateRequest): Future[Response] =
  bank.ask(replyTo => request.toCommand(id, replyTo))

After that, we need to invoke this method in an onSuccess directive like last time, and return an HTTP response in kind:

// instead of TODO 3
onSuccess(updateBankAccount(id, request)) {
  case BankAccountBalanceUpdatedResponse(Success(account)) =>
    complete(account)
  case BankAccountBalanceUpdatedResponse(Failure(ex)) =>
    complete(StatusCodes.BadRequest, FailureResponse(s"${ex.getMessage}"))
}

Our routes, therefore, will turn into this:

  val routes =
    pathPrefix("bank") {
      pathEndOrSingleSlash {
        post {
          // parse the payload
          entity(as[BankAccountCreationRequest]) { request =>
            onSuccess(createBankAccount(request)) {
              // send back an HTTP response
              case BankAccountCreatedResponse(id) =>
                respondWithHeader(Location(s"/bank/$id")) {
                  complete(StatusCodes.Created)
                }
            }
          }
        }
      } ~
      path(Segment) { id =>
        get {
          onSuccess(getBankAccount(id)) {
            case GetBankAccountResponse(Some(account)) =>
              complete(account)
            case GetBankAccountResponse(None) =>
              complete(StatusCodes.NotFound, FailureResponse(s"Bank account $id cannot be found."))
          }
        } ~
        put {
          entity(as[BankAccountUpdateRequest]) { request =>
            onSuccess(updateBankAccount(id, request)) {
              case BankAccountBalanceUpdatedResponse(Success(account)) =>
                complete(account)
              case BankAccountBalanceUpdatedResponse(Failure(ex)) =>
                complete(StatusCodes.BadRequest, FailureResponse(s"${ex.getMessage}"))
            }
          }
        }
      }
    }

Please watch carefully where the ~ operator is present: this is an operator to chain routes. Every HTTP request is matched by the routes of the server in turn; if the HTTP request was not matched by the current route, it will try the next one through the ~ operator.

Testing time!

Testing the HTTP Server

Much like we did earlier with testing the entire actor interaction, we’ll also create a standalone application that will spin up an ActorSystem, create the bank actor, and start a new HTTP server based on it.

For the ActorSystem, we need to be able to retrieve the Bank actor from inside of it, so we’ll need to send it a message and expect a response:

import akka.actor.typed.{ActorRef, ActorSystem, Behavior}
import akka.actor.typed.scaladsl.Behaviors
import com.rockthejvm.bank.actors.Bank
import com.rockthejvm.bank.actors.PersistentBankAccount.Command
import akka.actor.typed.scaladsl.AskPattern._
import akka.http.scaladsl.Http
import akka.util.Timeout
import com.rockthejvm.bank.http.BankRouter

import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.duration._
import scala.util.{Success, Failure}

object BankApp {
  trait RootCommand
  case class RetrieveBankActor(replyTo: ActorRef[ActorRef[Command]]) extends RootCommand

  val rootBehavior: Behavior[RootCommand] = Behaviors.setup { context =>
    val bankActor = context.spawn(Bank(), "bank")
    Behaviors.receiveMessage {
      case RetrieveBankActor(replyTo) =>
        replyTo ! bankActor
        Behaviors.same
    }
  }

  // continue here
}

Starting the HTTP server based on the bank actor will need some dedicated code as well:

def startHttpServer(bank: ActorRef[Command])(implicit system: ActorSystem[_]): Unit = {
  implicit val ec: ExecutionContext = system.executionContext
  val router = new BankRouter(bank)
  val routes = router.routes

  // start the server
  val httpBindingFuture = Http().newServerAt("localhost", 8080).bind(routes)

  // manage the server binding
  httpBindingFuture.onComplete {
    case Success(binding) =>
      val address = binding.localAddress
      system.log.info(s"Server online at http://${address.getHostString}:${address.getPort}")
    case Failure(ex) =>
      system.log.error(s"Failed to bind HTTP server, because: $ex")
      system.terminate()
  }
}

And in the main method, we now need to bring all pieces together:

  def main(args: Array[String]): Unit = {
    implicit val system: ActorSystem[RootCommand] = ActorSystem(rootBehavior, "BankSystem")
    implicit val timeout: Timeout = Timeout(5.seconds)
    implicit val ec: ExecutionContext = system.executionContext

    // using the ask pattern again
    val bankActorFuture: Future[ActorRef[Command]] = system.ask(replyTo => RetrieveBankActor(replyTo))
    bankActorFuture.foreach(startHttpServer)
  }

Run the application and have fun with the endpoints now, our application should be complete!