Как сделать WaitAll с Akka.Net?

Question

У меня есть иерархия актеров в Akka.Net, и мне интересно, правильно ли я выбрал способ сделать что-то или есть лучшие / более простые способы добиться того, чего я хочу.

Мой конкретный пример заключается в том, что я создаю актера User в ответ на вход пользователя в систему, и при создании этого актера мне нужны две части данных, чтобы завершить конструирование актера.

Если бы это был обычный код.NET, у меня могло бы быть что-то вроде следующего...

public Task<User> LoadUserAsync (string username)
{
  IProfileService profileService = ...;
  IMessageService messageService = ...;

  var loadProfileTask = profileService.GetUserProfileAsync(username);
  var loadMessagesTask = messageService.GetMessagesAsync(username);

  Task.WaitAll(loadProfileTask, loadMessagesTask);

  // Now construct the user from the result of both tasks
  var user = new User
  {
    Profile = loadProfileTask.Result,
    Messages = loadMessagesTask.Result
  }

  return Task.FromResult(user);
}

Здесь я использую WaitAll, чтобы дождаться завершения подчиненных задач и позволить им запускаться одновременно.

Мой вопрос - если бы я хотел сделать то же самое в Akka.Net, было бы следующее наиболее подходящим способом сделать это? Наглядно я создал следующее...

Когда я создаю свой пользовательский актер, я создаю (временный) пользовательский загрузчик, чья работа состоит в том, чтобы получить полную информацию о пользователе, вызывая актера профиля и актера сообщений. Конечные акторы, которые получают данные, следующие:

public class UserProfileLoader : ReceiveActor
{
    public UserProfileLoader()
    {
        Receive<LoadUserRequest>(msg =>
        {
            // Load the user profile from somewhere
            var profile = new UserProfile();

            // And respond to the Sender
            Sender.Tell(profile);
            Self.Tell(PoisonPill.Instance);
        });
    }
}

public class UserMessagesLoader : ReceiveActor
{
    public UserMessagesLoader()
    {
        Receive<LoadUserRequest>(msg =>
        {
            // Load the messages from somewhere
            var messages = new List<Message>();

            // And respond to the Sender
            Sender.Tell(messages);
            Self.Tell(PoisonPill.Instance);
        });
    }
}

На самом деле не имеет значения, откуда они получают данные для этого обсуждения, но оба просто отвечают на запрос, возвращая некоторые данные.

Тогда у меня есть актер, который координирует двух актеров сбора данных...

public class UserLoaderActor : ReceiveActor
{
    public UserLoaderActor()
    {
        Receive<LoadUserRequest>(msg => LoadProfileAndMessages(msg));
        Receive<UserProfile>(msg =>
        {
            _profile = msg;
            FinishIfPossible();
        });

        Receive<List<Message>>(msg =>
        {
            _messages = msg;
            FinishIfPossible();
        });
    }

    private void LoadProfileAndMessages(LoadUserRequest msg)
    {
        _originalSender = Sender;
        Context.ActorOf<UserProfileLoader>().Tell(msg);
        Context.ActorOf<UserMessagesLoader>().Tell(msg);
    }

    private void FinishIfPossible()
    {
        if ((null != _messages) && (null != _profile))
        {
            _originalSender.Tell(new LoadUserResponse(_profile, _messages));
            Self.Tell(PoisonPill.Instance);
        }
    }

    private IActorRef _originalSender;
    private UserProfile _profile;
    private List<Message> _messages;
}

Это просто создает двух подчиненных акторов, отправляет им сообщение для взлома, а затем ждет, пока оба ответят, перед отправкой обратно всех данных, которые были собраны исходному запрашивающему.

Итак, кажется ли это разумным способом скоординировать два разных ответа, чтобы объединить их? Есть ли более простой способ сделать это, чем сделать это самому?

Заранее спасибо за ваши ответы!

Answer 1

Спасибо, ребята, так что теперь я значительно упростил актера в следующем, основываясь на предложениях Роджера и Джеффа...

public class TaskBasedUserLoader : ReceiveActor
{
    public TaskBasedUserLoader()
    {
        Receive<LoadUserRequest>(msg => LoadProfileAndMessages(msg));
    }

    private void LoadProfileAndMessages(LoadUserRequest msg)
    {
        var originalSender = Sender;
        var loadPreferences = this.LoadProfile(msg.UserId);
        var loadMessages = this.LoadMessages(msg.UserId);

        Task.WhenAll(loadPreferences, loadMessages)
            .ContinueWith(t => new UserLoadedResponse(loadPreferences.Result, loadMessages.Result), 
                TaskContinuationOptions.AttachedToParent & TaskContinuationOptions.ExecuteSynchronously)
            .PipeTo(originalSender);
    }

    private Task<UserProfile> LoadProfile(string userId)
    {
        return Task.FromResult(new UserProfile { UserId = userId });
    }

    private Task<List<Message>> LoadMessages(string userId)
    {
        return Task.FromResult(new List<Message>());
    }
}

Методы LoadProfile и LoadMessages в конечном итоге вызовут репозиторий для получения данных, но сейчас у меня есть краткий способ сделать то, что я хотел.

Еще раз спасибо!

Answer 2

ИМХО, это правильный процесс, так как вы разветвляете действие и затем присоединяетесь к нему.

Кстати, вы могли бы использовать this.Self.GracefulStop(new TimeSpan(1)); вместо отправки отравленной таблетки.

Answer 3

Вы можете использовать комбинацию Ask, WhenAll и PipeTo:

var task1 = actor1.Ask<Result1>(request1);
var task2 = actor2.Ask<Result2>(request2);

Task.WhenAll(task1, task2)
    .ContinueWith(_ => new Result3(task1.Result, task2.Result))
    .PipeTo(Self);

...

Receive<Result3>(msg => { ... });