When building distributed services one of the core concepts is domain message serialization. Since Akka.NET core concept is message passing between actors located in different processes, efficient message serialization plays a big role in making the solution performant and easy to maintain.
This article requires that reader is familar with core concept of Akka.NET serialization (see https://getakka.net/articles/serialization/serialization.html) and ProtoBuf-Net library (see https://github.com/protobuf-net/protobuf-net).
Main reason why I have decided to use protobuf-net and not Google .proto files, was the freedom to define my messages as read-only record types. Akka.NET strongly advice to make the domain messages as immutable (here is a great article https://petabridge.com/blog/how-to-design-akkadotnet-domain-messages/), and as we well know, when using Google’s .proto files to generate C# classes, such messages are not immutable, therefore to follow best practices in Akka.NET, one would have to create both .proto files and then manually map to our domain messages. Tedious task, that would require spending hours of writing useless lines of code mapping field by field both when we serialize and deserialize, with a likelyhood of introducing mapping errors. Choosing protobuf-net, we only write domain messages and register them with 1 line of code in the ActorSystem, and whole serialization will be handled for us.
Working example of applying this serialization approach, can be found on github (link below). In this example we have 2 actors AccountManagerActor and AccountVerificationActor (I know, not the best names, but since its second hardest thing in programming I leave it to you to come up with better names). Each actor is hosted in a seperate service, AccountManagerActor in Service1 and AccountVerificationActor in Service2. Both services form an Akka.NET Cluster. AccountVerificationActor is the one that communicates with AccountManagerActor via Akka.NET Remoting calls which are invisible to the user, but will allow us to see that the serialization is in fact used when messages are send to remote actor.
We start with an implementation of a generic abstract class ProtobufMessageSerializer that extends Akka.Serialization.SerializerWithStringManifest serializer class. This class is where the serialization and deserialization is happening, therefore if we would like to tweek the protobuf-net serialization/deserialization process, this is the place.
public abstract class ProtobufMessageSerializer<TMessageSerializer> : SerializerWithStringManifest
where TMessageSerializer : ProtobufMessageSerializer<TMessageSerializer>
{
private static readonly Dictionary<Type, string> manifest = new();
private static readonly Dictionary<Type, Func<object, byte[]>> serializers = new();
private static readonly Dictionary<string, Func<byte[], object>> deserializers = new();
protected ProtobufMessageSerializer(int identifier, ExtendedActorSystem system) : base(system)
{
Identifier = identifier;
}
/// <summary>
/// Unique value greater than 100 as [0-100] is reserved for Akka.NET System serializers.
/// </summary>
public override int Identifier { get; }
protected static void Add<T>(string manifestKey) where T : class
{
if (!manifest.TryAdd(typeof(T), manifestKey))
throw new ArgumentException($"MessageSerializer Manifest already added: {manifestKey}->{typeof(T).Name}", nameof(manifestKey));
if (!serializers.TryAdd(typeof(T), Serialize))
throw new ArgumentException($"MessageSerializer Type already added: {manifestKey}->{typeof(T).Name}", nameof(manifestKey));
deserializers[manifestKey] = Deserialize<T>;
}
public static object Deserialize<T>(byte[] bytes) where T : class
=> ProtoBuf.Serializer.Deserialize<T>(new ReadOnlySpan<byte>(bytes));
public static byte[] Serialize(object obj)
{
using var memoryStream = new MemoryStream();
ProtoBuf.Serializer.Serialize(memoryStream, obj);
return memoryStream.ToArray();
}
public static bool TryGetManifest<T>([MaybeNullWhen(false)] out string manifestKey)
=> manifest.TryGetValue(typeof(T), out manifestKey);
public static bool TryGetManifest(Type objectType, [MaybeNullWhen(false)] out string manifestKey)
=> manifest.TryGetValue(objectType, out manifestKey);
public static bool TryGetSerializer(Type objectType, [MaybeNullWhen(false)] out Func<object, byte[]> serializer)
=> serializers.TryGetValue(objectType, out serializer);
public static bool TryGetSerializer<T>([MaybeNullWhen(false)] out Func<object, byte[]> serializer)
=> serializers.TryGetValue(typeof(T), out serializer);
public static bool TryGetDeserializer(string manifestKey, [MaybeNullWhen(false)] out Func<byte[], object> deserializer)
=> deserializers.TryGetValue(manifestKey, out deserializer);
public static string GetManifest<T>()
{
if (manifest.TryGetValue(typeof(T), out var key))
return key;
throw new ArgumentOutOfRangeException("{T}", $"Unsupported message type [{typeof(T)}]");
}
public override string Manifest(object o)
{
if (TryGetManifest(o.GetType(), out var key))
return key;
throw new ArgumentOutOfRangeException(nameof(o), $"Unsupported message type [{o.GetType()}]");
}
public override object FromBinary(byte[] bytes, string manifest)
{
if (TryGetDeserializer(manifest, out var deserializer))
return deserializer(bytes);
throw new ArgumentOutOfRangeException(nameof(manifest), $"Unsupported message manifest [{manifest}]");
}
public override byte[] ToBinary(object obj)
{
if (TryGetSerializer(obj.GetType(), out var serializer))
return serializer(obj);
throw new ArgumentOutOfRangeException(nameof(obj), $"Unsupported message type [{obj.GetType()}]");
}
}
Next step is to implement a message that will be used in actor communication and the concrete serializer class that will extend our abstract serializer. This is the place where we register the messages that are exchanged between remote actors and that we want our serializer to handle, and specify the id underd which Akka.NET will have this given serializer registered.
/// <summary>
/// Marker interface for all messages in the app domain.
/// Used by Akka.NET to select the correct serializer.
/// </summary>
public interface IProtocolMember {}
[ProtoContract]
public sealed record AccountCreated : IProtocolMember
{
[ProtoMember(1)] public required string AccountId { get; init; }
}
[ProtoContract]
public sealed record CreateAccount : IProtocolMember
{
[ProtoMember(1)] public required string AccountId { get; init; }
[ProtoMember(2)] public required string Name { get; init; }
}
public sealed class MessageSerializer : ProtobufMessageSerializer<MessageSerializer>
{
//Unique Id of our serializer, required by Akka.NET
public const int Id = 1000;
static MessageSerializer()
{
Add<CreateAccount>("ca");
Add<AccountCreated>("ac");
}
public MessageSerializer(ExtendedActorSystem system) : base(Id, system) { }
}
Having defined our protocol messages and serializer, the last thing to do before Akka.NET ActorSystem will be able to use it, is to register it with the system. To accomplish it, we use WithCustomSerializer extension method defined in Akka.Hosting package.
builder.Services.AddAkka("smaple-app", builder =>
{
builder
.WithCustomSerializer(
"account-messages",
new[] { typeof(IProtocolMember) },
system => new MessageSerializer(system));
});
That’s it! Full working example can be found on github (https://github.com/rafalpiotrowski/AkkaProtoBufMessageSerialization)
Top comments (0)