Akka.NET #4: Connecting Actor Model with endpoints

As you’ve probably seen in previous posts about Akka.NET, actor model is no rocket science and is really easy and fun to start with. However entire actor model may seem like closed and hermetic ecosystem, today I’ll show you how to poke actors in way that’ll make them poke back.

As for now almost entire logic in my project happens inside of an actor. Exceptions are few helpers, API endpoints layer, entities which are being represented by it’s own actors and obviously connectors in form of Dispatchers and Enquirers which’re sending commands and queries to entire actor systems.

I’m using NancyFX so my endpoint are all about HTTP Requests and Responses but everything I’ll write down below is applicable with MVC, desktop or mobile apps or even cases where you need to use actor model only in small part of your application.

Of course it’s not the best way and there are other ones, I’ve chosen this because it’s modular and easy to extend whenever I want. Only parts that’s strictly “Akka” themed are Tell, Forward and Ask methods, which are basic methods of sending messages to actors, everything else could be done in some other way.

Commands and Queries

Almost every action that can be done in interface of almost every application can be described as “I want you to do something” or “I want you to give me something”. Or in common technical terms – Commands and Queries. If you want to know more about them just google for CQRS which is principle that separates those two and gives much attention to them.

I’m not going all-the-way CQRS but I’ve separated commands of queries because their behavior is really different. For example – when user rates something, by executing RateSomething command, we could wait for confirmation that everything was executed successfully or in case of my project just sen some messages to certain actors and return HttpStatusCode.Accepted which means request was accepted and is being processed, simple rating of some link isn’t so important to wait for confirmation.

That means with commands we can tell our actors to do something and just return response that it’s done, show change in UI even when message that we’ve sent is still in some message queue or is being processed by some fancy service. We could want confirmation or not, that’s our decision to make.

Queries are a bit different because we need results and most of the time we’ll need them ASAP and that means we need to wait for them. In actor model that means sending message to some one or more actors and waiting for it to pass the queue and reply, sometimes it could take some time but we can be interested in for example 10 results, send our query to hundreds of actors and wait for first 10 results. As long as we’ll remember that query should never ever change state of anything in our application we can grab those 10 results and go away, other actors will process their queries and send responses but we won’t use them in any way.

Commands

As I’ve mentioned commands are basically “do something” requests. That means they’re changing state of application in my case this being actor’s state. Every command in Me2.0 have a starting point in some kind of entity. For now i have three of them – user, content and tag. Entities are being binded from http requests bodies (commands will rarely be made using GET requests).

Every existing entity in my project implements IHaveDispatchers interface which is responsible for attaching Dispatchers to entity and executing them and returning appropriate status code.

public interface IHaveDispatchers<T> : IEntity
{
    T With(IDispatch<T> dispatcher);
    T With(IEnumerable<IDispatch<T>> dispatchers);
    T WithSpecific(IEnumerable<IDispatch<T>> dispatchers, params string[] internalNames);

    HttpResult<T> DispatchAll(string userName);
}

public interface IHaveDispatchers<T> : IEntity

{

T With(IDispatch<T> dispatcher);

T With(IEnumerable<IDispatch<T>> dispatchers);

T WithSpecific(IEnumerable<IDispatch<T>> dispatchers, params string[] internalNames);

HttpResult<T> DispatchAll(string userName);

}

Dispatchers are also present in Akka.NET but this mechanism is mine, I’ll probably change their name to other as soon as I’ll have any good idea for that. They are simple objects that’re just firing some messages to actors without returning anything (for now, I’m planning on implementing DispatchAndConfirm method in the future).

public interface IDispatch<T> : IHaveInternalName
{
    void Dispatch(T item, string userName);
}

public interface IDispatch<T> : IHaveInternalName

{

void Dispatch(T item, string userName);

}

What’s the point of that? As I’ve written before – I want absolute modularity of my application. For now I have for example RateContentDispatcher which is being attached to Content entity during POST request when we click any button responsible for rating some content.

For now this dispatcher is responsible for telling any interested party to rate that content. That means sending message to UserActor with information “You need to rate this content for x points” and to ContentActor information “You’re being rated by this user for x points” – every interested actor should be notified and should change it’s state, from business point of view this command is not significant enough to check if that happened but we could do this in our dispatcher.

public class RateContentDispatcher : DispatcherBase<ContentEntity>
{
    public static readonly string Name = "RateContent";

    public RateContentDispatcher() : base()
    { }

    public override void Dispatch(ContentEntity item, string userName)
    {
        var command = new RateContentCommand(item.Uri, userName, item.Rating);
        ActorModel.UsersManagerActorRef.Tell(command);
        ActorModel.ContentManagerActorRef.Tell(command);
    }
}

public class RateContentDispatcher : DispatcherBase<ContentEntity>

{

public static readonly string Name = "RateContent";

public RateContentDispatcher() : base()

{ }

public override void Dispatch(ContentEntity item, string userName)

{

var command = new RateContentCommand(item.Uri, userName, item.Rating);

ActorModel.UsersManagerActorRef.Tell(command);

ActorModel.ContentManagerActorRef.Tell(command);

}

Imagine simple and possible situation. In the near future I would want to add some kind of statistics actor that would like to gather some information about things that going on in our application. It could be one actor instance because losing some stats data is acceptable. Using dispatchers in way I’m doing that it means I could just create some AddStatsDispatcher class and just attach it where I would want them. No changes to existing logic would be needed during attaching new dispatchers or deleting them and that’s the point. And my POST handlers looks really simple and clear this way.

Post["/"] = p =>
{
    var content = this.Bind<ContentEntity>();
    return Response.AsJson(content
        .WithSpecific(dispatchers,
            AddContentDispatcher.Name)
        .DispatchAll(Context.CurrentUser.UserName));
};

Post["/"] = p =>

{

var content = this.Bind<ContentEntity>();

return Response.AsJson(content

.WithSpecific(dispatchers,

AddContentDispatcher.Name)

.DispatchAll(Context.CurrentUser.UserName));

};

Queries

While command could be attached to some entities acquired by binding request data to model tell something what’s need to be done and not worry about that anymore, queries are something different because we need some kind of results. I didn’t wanted to attach my queries to any entities because they supposed to return those and not be dependent on them, that’s why I created Enquirers which are nothing more than containers for query objects which in turn are similar to Dispatcher but they return stuff.

public interface IEnquire<T> where T : IEntity
{
    TimeSpan AcceptableTimeout { get; }
    IEnquire<T> QueryFor(IQuery<T> query);
    IEnquire<T> QueryForAll(params IQuery<T>[] queries);
    IEnquire<T> QueryForAll(IEnumerable<IQuery<T>> queries);

    IEnquire<T> WaitFor(TimeSpan timeout);
    HttpResult<IEnumerable<T>> Execute();
}

public interface IQuery<T> where T : IEntity
{ 
    IEnumerable<T> Execute(IEnquire<T> enquirer);
}

public interface IEnquire<T> where T : IEntity

{

TimeSpan AcceptableTimeout { get; }

IEnquire<T> QueryFor(IQuery<T> query);

IEnquire<T> QueryForAll(params IQuery<T>[] queries);

IEnquire<T> QueryForAll(IEnumerable<IQuery<T>> queries);

IEnquire<T> WaitFor(TimeSpan timeout);

HttpResult<IEnumerable<T>> Execute();

}

public interface IQuery<T> where T : IEntity

{

IEnumerable<T> Execute(IEnquire<T> enquirer);

}

That means I can have few queries inside one Enquirer and Aggregate results inside of it, sometime it will be simple merge, sometimes summing and sometimes we need to aggregate data stored in two different places or remove some inconsistencies. Execute method in IQuery just asks actors for data and returns results and Execute method in Enquirer invokes Execute in it’s IQuery objects and aggregates results as we see fit.

Most basic Enquirer I have just returns all content that user saved. As content is separate entity with it’s own global state like average rating, tags from many users etc., not every data about content is stored within UserActor state, to be honest User don’t know anything more than how high was content rated, with what was it tagged and so on but everything in current user scope. So first we need to fetch every content that we’ve added to our actors state. We’ll use ordinary ContentEnquirer that just returns everything it gets as response, everything is in generic base class and it’s just “grab everything and wrap into some kind of response” but we don’t need anything more at the moment. Let’s focus on our query, which is simple too.

public class GetUserContentQuery : IQuery<ContentEntity>
{
    private readonly string userName;
    public GetUserContentQuery(string userName)
    {
        this.userName = userName;
    }
    public IEnumerable<ContentEntity> Execute(IEnquire<ContentEntity> enquirer)
    {
        if (string.IsNullOrEmpty(userName))
            throw new ArgumentException("userName parameter in GetUserContentQuery should not be null or empty");

        var result = ActorModel.UsersManagerActorRef.Ask(new GetUserContentQueryMessage(userName), enquirer.AcceptableTimeout).Result as GetUserContentQueryResultMessage;

        return result == null ?
            Enumerable.Empty<ContentEntity>() :
            result.ContentWithTags.Select(cwt =>
                new ContentEntity
                {
                    Url = cwt.Key.ToString(),
                    Tags = cwt.Value.Select(v => new TagDTO(v, true)).ToList()
                }
            );
    }
}

public class GetUserContentQuery : IQuery<ContentEntity>

{

private readonly string userName;

public GetUserContentQuery(string userName)

{

this.userName = userName;

}

public IEnumerable<ContentEntity> Execute(IEnquire<ContentEntity> enquirer)

{

if (string.IsNullOrEmpty(userName))

throw new ArgumentException("userName parameter in GetUserContentQuery should not be null or empty");

var result = ActorModel.UsersManagerActorRef.Ask(new GetUserContentQueryMessage(userName), enquirer.AcceptableTimeout).Result as GetUserContentQueryResultMessage;

return result == null ?

Enumerable.Empty<ContentEntity>() :

result.ContentWithTags.Select(cwt =>

new ContentEntity

{

Url = cwt.Key.ToString(),

Tags = cwt.Value.Select(v => new TagDTO(v, true)).ToList()

}

);

}

I haven’t made it async yet so it’s kind of ugly but it shows what it should do and I will fix it someday. Key part is in line 13 which I’ve marked. It’s Ask method which is similar to Tell method but it returns Task<object> with response if we receive any.

At this moment temporary Sender is being created and our message will be sent. As for message I’ve grouped them in pairs of QueryMessages which are our query intent like “GetUserContent” plus some data needed for delivering message (like user name) and/or some query details. Second message in pair is QueryResultMessage which is container for response. That’s how they look like in this example.

public class GetUserContentQueryMessage : IHaveUserName
{
    public string UserName { get; private set; }
    public GetUserContentQueryMessage(string userName)
    {
        UserName = userName;
    }
}

public class GetUserContentQueryResultMessage
{
    public IReadOnlyDictionary<Uri, HashSet<string>> ContentWithTags { get; private set; }

    internal GetUserContentQueryResultMessage(ContentContainer usersContentsContainer)
    {
        ContentWithTags = usersContentsContainer.Values.ToDictionary(c => c.Uri, c => c.Tags);
    }
}

public class GetUserContentQueryMessage : IHaveUserName

{

public string UserName { get; private set; }

public GetUserContentQueryMessage(string userName)

{

UserName = userName;

}

public class GetUserContentQueryResultMessage

{

public IReadOnlyDictionary<Uri, HashSet<string>> ContentWithTags { get; private set; }

internal GetUserContentQueryResultMessage(ContentContainer usersContentsContainer)

{

ContentWithTags = usersContentsContainer.Values.ToDictionary(c => c.Uri, c => c.Tags);

}

Firs we’re sending our message to UsersManagerActor. It Forwards every message that implements IHaveUserName interface to target actor. Forward method is yet another way to send message, this one preserves original Sender so we’ll reply back and don’t need to pass any more data about message origination point.

public UsersManagerActor() : base()
{
    //..
    Receive<IHaveUserName>(msg => Context.Child(msg.UserName).Forward(msg));
}

public UsersManagerActor() : base()

{

//..

Receive<IHaveUserName>(msg => Context.Child(msg.UserName).Forward(msg));

}

At the destination, in our UserActor we’ve just receiving our QueryMessage, building QueryResultMessage and sending it back to Sender (Command method is just like Receive but it’s being used by PersistentActors which we’ll cover soon).

public UserActor(string authenthicationType, string id) : base()
{
    //...
    Command<GetUserContentQueryMessage>(msg => Sender.Tell(new GetUserContentQueryResultMessage(ActorState.Contents)));
    //...
}

public UserActor(string authenthicationType, string id) : base()

{

//...

Command<GetUserContentQueryMessage>(msg => Sender.Tell(new GetUserContentQueryResultMessage(ActorState.Contents)));

//...

}

After all that we’re ready to receive results in our query object and cast it to desired type of result message, project it into collection of entities and return it with our endpoint. In my scenario I’m fetching this basic data just after page load and when I’ve got some basics I’m querying for more details like titles, average ratings etc. individually for each content, then it’s necessary to query for data both User and Content actors and just aggregate results into desired shape.

Summary

Both querying and commanding our actors are pretty simple. We could ask for great things or small, granular favors. Everything depends on us and our business needs. I wanted to keep thing modular and I’ve done it, I can introduce new entity and it’s set of actor even tommorow and it will not require touching of existing code even if new entity will be tightly connected to existing ones, same thing with deleting or turning off entire segment of application if I ever will want to do this.