Обновлять состояние актора только после того, как все события сохранятся

В методе приема постоянного актора я получаю группу событий, которые хочу сохранить, и только после того, как все события сохраняются, снова обновляю свое состояние. Как я могу это сделать?

def receive: Receive = {
  ...
  case NewEvents(events) =>
    persist(events) { singleEvent =>
      // Update state using this single event
    }
    // After every events are persisted, do one more thing
}

Обратите внимание, что вызов persist () не блокируется, поэтому я не могу вставить свой код сразу после этого.

Обновление: зачем мне это

Эти новые события поступают из внешней веб-службы. Моему постоянному актору необходимо сохранить в своем состоянии последний идентификатор события, который будет использоваться для последующего вызова ws, когда он получит команду. Дело в том, что эти команды могут приходить одновременно, поэтому мне нужна какая-то система блокировки:

  • Полученная команда вызова ws: отложить следующие команды, пока не завершится эта (то есть, в общем, логическое значение)
  • Полученные ответы от ws: сохранить их, обновить состояние и сохранить последний идентификатор, выполнить еще один, единственный вызов ws для всех команд, находящихся в тайнике (я сохраняю отправителей команд, чтобы они могли ответьте на них все после того, как закончите) в противном случае больше не прячьте команды.

scala akka akka-persistence
person Dimitri    schedule 18.12.2014    source источник
comment
defer работает для ваших целей?   -  person Ryan    schedule 18.12.2014
comment
Выглядит многообещающе, я попробую это. Спасибо!   -  person Dimitri    schedule 18.12.2014
comment
Могу я спросить, в чем причина вашей потребности? Может быть, есть лучший способ достичь этого, если вы объясните свою конечную цель   -  person Diego Martinoia    schedule 19.12.2014
comment
@DiegoMartinoia Я обновил свой ответ :)   -  person Dimitri    schedule 20.12.2014
comment
Вы также можете написать свой собственный плагин Persistence Journal в качестве оболочки для существующего, а затем делать все, что вам нужно, после того, как вызовы синхронизации завершатся оттуда.   -  person Diego Martinoia    schedule 21.12.2014

Ответы (1)


arrow_upward
1
arrow_downward

Я еще не пробовал defer, моим первоначальным решением было отправить себе PersistEventsDone сообщение. Это работает, потому что метод persist будет хранить все входящие сообщения до тех пор, пока не будут выполнены все обработчики событий. Если в процессе появилась другая команда, не имеет значения, до или после PersistEventsDone:

def receive: Receive = {
  ...
  case PersistEventsDone =>
    ...
  case NewEvents(events) =>
    persist(events) { singleEvent =>
      // Update state using this single event
    }
    self ! PersistEventsDone
}

defer в моем случае немного странный, потому что для него требуется событие, которое мне не нужно. Но все равно выглядит более естественным, чем мое решение.

person Dimitri    schedule 19.12.2014