Introduction: Event Sourcing
In most modern games, it is conventional wisdom that you should save your progress regularly, lest you take a wrong turn and get mauled:
Software is no different. If you’re running an actor system in memory, you risk losing state if anything happens to the actors or to the whole actor system.
It is thus important to save the state of your actors, but how?
In message-based systems such as Akka .NET, a popular approach towards recovery is to save messages as they arrive, and in case of failure, simply handle them again in the same order to restore the last state. This is known as event sourcing.
Akka .NET provides event sourcing support thanks to the Akka.Persistence module, as we shall see shortly.
The source code for this article is available at the Gigi Labs BitBucket repository.
Chess Scenario
Chess is a great example with which to demonstrate event sourcing because a chess game consists of a sequential set of moves which can be represented using a specific notation. We can save these moves and later replay the game.
It is also very easy to draw a chess board in a console application. It’s just an 8×8 grid with the pieces on it. We can represent the various chess pieces using different letters, and use uppercase and lowercase (instead of white and black) to distinguish between the two players’ pieces, as GNU Chess does:
Chess can get quite complex and I really don’t want to get lost in the details (since this article is about Akka.Persistence), so we’ll make a number of assumptions as follows to keep things simple:
- No validation. Pieces can be moved anywhere on the board.
- Both players use the same console window and take it in turns.
- No game state (i.e. you can never win or lose).
- Input will be of the format:
move <from> to <to>, for examplemove e2 to e4.
In other words, we’re not really building a chess game. We’re just emulating the board and the movement so that we can store moves and replay them later.
Prerequisites
To follow along, install the following NuGet packages:
Install-Package Akka Install-Package Akka.Persistence -pre Install-Package Akka.Persistence.SqlServer -pre
Akka.Persistence seems to be prerelease at the time of writing this article, so you will need the -pre flag.
Akka.Persistence is very flexible in terms of where messages are saved. In this example we’re going to use SQL Server, but there are a whole load of storage implementations. Just look up “Akka.Persistence” in NuGet and you’ll see the available options:
System Overview
We have a ChessGameActor that holds the game state (i.e. the chess board). I was originally going to use a string[] for this, but since we need to update individual characters, the immutable nature of strings becomes a problem. We need to use a 2-dimensional char array instead.
public class ChessGameActor : ReceiveActor
{
private Guid gameId;
private IActorRef rendererActor;
private char[][] chessBoard = new char[][]
{
"rnbqkbnr".ToCharArray(),
"pppppppp".ToCharArray(),
" ".ToCharArray(),
" ".ToCharArray(),
" ".ToCharArray(),
" ".ToCharArray(),
"PPPPPPPP".ToCharArray(),
"RNBQKBNR".ToCharArray()
};
// ...
}
We also have a ChessBoardDrawingActor responsible for actually drawing the chess board. The ChessGameActor has a reference to it so that it can ask it to redraw the board when someone moves a piece.
The details of how ChessBoardDrawingActor is implemented are omitted for brevity (refer to the source code if you need it), but it basically just handles DrawChessBoardMessages coming from the ChessGameActor:
public class ChessBoardDrawingActor : ReceiveActor
{
public ChessBoardDrawingActor()
{
this.Receive<DrawChessBoardMessage>(m => Handle(m));
}
public void Handle(DrawChessBoardMessage message)
{
Console.Clear();
var chessBoard = message.ChessBoard;
// ...
}
Although you technically could do this from the ChessGameActor itself, I consider it good practice to separate state/logic from presentation. Reminiscent of the MVC pattern, this makes it easy to support various output devices (e.g. GUI window, web, mobile, etc) without having to change the core of your game.
The DrawChessBoardMessage is simply a copy of the chessboard:
public class DrawChessBoardMessage
{
public char[][] ChessBoard { get; }
public DrawChessBoardMessage(char[][] chessBoard)
{
this.ChessBoard = chessBoard;
}
}
Although we could micro-optimise this by sending a diff instead (i.e. old position to erase, and new position to draw) as we do in the Akka.Remote multiplayer game example, the data here is so small as to carry negligible overhead. Besides, it’s common practice in games to just redraw everything (which may not be the fastest approach, but complex environments make tracking changes impossible).
The main program is responsible for creating the actor system, along with these two actors:
static void Main(string[] args)
{
Console.Title = "Akka .NET Persistence Chess Example";
using (var actorSystem = ActorSystem.Create("Chess"))
{
var drawingProps = Props.Create<ChessBoardDrawingActor>();
var drawingActor = actorSystem.ActorOf(drawingProps, "DrawingActor");
Guid gameId = Guid.Parse("F56079D3-4625-409A-B734-C9BDEBA6D7FA");
var gameProps = Props.Create<ChessGameActor>(gameId, drawingActor);
var gameActor = actorSystem.ActorOf(gameProps, "GameActor");
HandleInput(gameActor);
Console.ReadLine();
}
}
The input handling logic expects to receives moves in the format move <from> to <to>; once it extracts the from and to locations, it sends a MoveMessage to the ChessGameActor.
static void HandleInput(IActorRef chessGameActor)
{
string input = string.Empty;
while (input != null) // quit on Ctrl+Z
{
input = Console.ReadLine();
var tokens = input.Split();
switch (tokens[0]) // check first word
{
case "move": // e.g. move e2 to e4
{
string from = tokens[1];
string to = tokens[3];
var message = new MoveMessage(from, to);
chessGameActor.Tell(message);
}
break;
default:
Console.WriteLine("Invalid command.");
break;
}
}
}
In fact, a MoveMessage is simply a combination of from and to locations:
public class MoveMessage
{
public string From { get; }
public string To { get; }
public MoveMessage(string from, string to)
{
this.From = from;
this.To = to;
}
public override string ToString()
{
return $"move {this.From} to {this.To}";
}
}
However, these locations are still in the format entered by the user (e.g. e4). When the GameActor receives a MoveMessage, it must first translate the locations into indices in the 2-dimensional array that we’re using as a chess board. This is done in a method called TranslateMove() which does some funky ASCII manipulation…
private Point TranslateMove(string move)
{
// e.g. e4: e is the column, and 4 is the row
char colChar = move[0];
char rowChar = move[1];
int col = colChar - 97;
int row = 8 - (rowChar - '0');
return new Point(col, row);
}
…and returns an instance of a Point class. Point is your typical 2D coordinate.
public class Point
{
public int X { get; }
public int Y { get; }
public Point(int x, int y)
{
this.X = x;
this.Y = y;
}
}
Once the GameActor translates these coordinates, it can update the state of the chess board, and send a DrawChessBoardMessage to the ChessBoardDrawingActor to redraw the chess board.
public void Handle(MoveMessage message)
{
var fromPoint = this.TranslateMove(message.From);
var toPoint = this.TranslateMove(message.To);
char piece = this.chessBoard[fromPoint.Y][fromPoint.X];
chessBoard[fromPoint.Y][fromPoint.X] = ' '; // erase old location
chessBoard[toPoint.Y][toPoint.X] = piece; // set new location
this.RedrawBoard();
}
private void RedrawBoard()
{
var drawMessage = new DrawChessBoardMessage(this.chessBoard);
this.rendererActor.Tell(drawMessage);
}
Saving Messages using Akka.Persistence Journaling
In order to be able to recover our actor’s state (in this case, replay chess games one move at a time), we need to store those MoveMessages as they arrive in our ChessGameActor. We can do this using the built-in functionality of Akka.Persistence.
The first thing we need to do is have our ChessGameActor inherit from ReceivePersistentActor (instead of ReceiveActor):
public class ChessGameActor : ReceivePersistentActor
When we do this, we will be required to provide a property called PersistenceId. Fortunately, we’re passing in a Guid called gameId to our actor, so we can use that:
public override string PersistenceId
{
get
{
return this.gameId.ToString("N");
}
}
public ChessGameActor(Guid gameId, IActorRef rendererActor)
{
this.gameId = gameId;
// ...
}
We’ll see what this is for in a minute. Let’s complete our constructor:
public ChessGameActor(Guid gameId, IActorRef rendererActor)
{
this.gameId = gameId;
this.rendererActor = rendererActor;
this.RedrawBoard();
this.Command<MoveMessage>(PersistAndHandle, null);
}
In the constructor, we store the game ID and a reference to the ChessBoardDrawingActor. We draw the initial board (before anyone has moved), and then we set up our message handling.
In a ReceivePersistentActor, we use Command<T>() instead of Receive<T>() to set up our message handlers. We can’t use Receive<T>() because of some ambiguity between base class methods. The null value passed in is similarly to prevent ambiguities between overloads.
In the PersistAndHandle() method, we call the built-in Persist() method to save the message and call a handling method after the save is successful:
public void PersistAndHandle(MoveMessage message)
{
Persist(message, persistedMessage => Handle(persistedMessage));
}
The Handle() method is the same one we’ve seen before that handles the MoveMessage.
We could have done all this in one step within the Command<T>() call, as you can see in Petabridge’s Akka.Persistence blog article. However, I’m not a big fan of doing a lot of logic in nested lambdas, as they can quickly get out of hand for non-trivial scenarios.
Now we just need a little configuration to tell Akka.Persistence where it should store the messages whenever we call Persist():
<?xml version="1.0" encoding="utf-8" ?>
<configuration>
<configSections>
<section name="akka" type="Akka.Configuration.Hocon.AkkaConfigurationSection, Akka" />
</configSections>
<startup>
<supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.6.1" />
</startup>
<akka>
<hocon>
<![CDATA[
akka.persistence
{
journal
{
plugin = "akka.persistence.journal.sql-server"
sql-server
{
class = "Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.SqlServer"
schema-name = dbo
auto-initialize = on
connection-string = "Data Source=.\\SQLEXPRESS;Database=AkkaPersistenceChess;Integrated Security=True"
}
}
}
]]>
</hocon>
</akka>
</configuration>
Let’s run the program and enter a few moves that we can later replay. These are the moves for fool’s mate:
move f2 to f3 move e7 to e5 move g2 to g4 move d8 to h4
Here is what it looks like, just before the last move:
If you check your database now, you’ll see a couple of tables. The EventJournal one has an entry for each move you played (i.e. for each message that was handled):
The values in the PersistenceId column match the game ID, which is what we provided in the ChessGameActor’s PersistenceId property. If we wanted to save the progress of a different game, we would pass in a different game ID (and thus PersistenceId) to our ChessGameActor.
Each PersistenceId should be unique across the across the actor system, and there should be only one instance of a persistence actor with that PersistenceId running at any given time. Doing otherwise would compromise the state saved in the database.
Recovering State
In our ChessGameActor’s constructor, we can use the Recover<T>() method to replay messages from the persistence store and recover our state (i.e. do event sourcing), before we begin receiving new messages.
public ChessGameActor(Guid gameId, IActorRef rendererActor)
{
this.gameId = gameId;
this.rendererActor = rendererActor;
this.RedrawBoard();
this.Recover<MoveMessage>(RecoverAndHandle, null);
this.Command<MoveMessage>(PersistAndHandle, null);
}
In this case, we’ll handle the recovered messages as normal, but we will also introduce an artificial delay so that the player can actually watch each move being replayed.
public void RecoverAndHandle(MoveMessage message)
{
Handle(message);
Thread.Sleep(2000);
}
If we run the game again now, we can watch the moves being replayed, and then we can continue playing where we left off.
Summary
In this article, we have learned how Akka.Persistence supports event sourcing. This is done as follows:
- Actors wanting to save messages should inherit from ReceivePersistentActor.
- They must supply a
PersistenceIdwhich is unique and which will be used to associate saved messages with this particular actor’s state (or that of any subsequent incarnations). - Use
Command<T>()instead ofReceive<T>()for message handling. - Use
Persist()to save messages before handling them. - Use
Recover()to replay messages until the actor’s last state is restored.
The particular approach we have seen is called journaling, and it is only one feature of Akka.Persistence. This may be enough for chess games that typically last for not more than 30-40 moves. But in many other use cases with large data flows, the journal may grow a lot and it can take a while to restore state. Akka.Persistence supports a snapshot feature to help mitigate this problem.