I have a simple test of Akka persistence with LevelDB. First app creates
10000 aggregates, second app updates them, third app logs them (via a
GetContent message). The good news is that the creation and updating seems
pretty snappy. In the logging (GetContent) app, I get the following error
(many times), but it does appear to complete the logging.
[DEBUG] [10/18/2014 11:35:45.462]
[log-persistent-actors-test-akka.persistence.dispatchers.default-replay-dispatcher-34]
[LocalActorRefProvider(akka://log-persistent-actors-test)] resolve of
unknown path [akka://create-persistent-actors-test/deadLetters] failed
The only difference I see between the three scenarios is that in the first
two cases a persist() call happens. Is this error benign, or should I be
sending a recover message before interacting normally with a persistent
actor?
Each app looks like this:
>>>
object LogPersistentActors extends App {
println("Log persistent actors app")
implicit val system = ActorSystem("log-persistent-actors-test")
val supervisor = system.actorOf(SamplePersistentActorSupervisor.props)
for (i <- 1 to 10000) {
val id = s"id-$i"
supervisor ! SamplePersistentActorSupervisor.GetContent(id)
}
}
<<<
The parent aggregate manager is doing this:
>>>>
override def receive: Receive = {
case SamplePersistentActorSupervisor.CreateAggregate(id, content) => {
processAggregateCommand(id, SamplePersistentActor.Initialize(content))
}
case SamplePersistentActorSupervisor.UpdateContent(id, content) => {
processAggregateCommand(id, SamplePersistentActor.ChangeContent(
content))
}
case SamplePersistentActorSupervisor.GetContent(id) => {
processAggregateCommand(id, SamplePersistentActor.GetContent)
}
}
def processAggregateCommand(aggregateId: String, command:
SamplePersistentActor.Command) = {
val maybeChild = context child aggregateId
maybeChild match {
case Some(child) => child forward command
case None => {
val child = createAggregate(aggregateId)
child forward command
}
}
}
private def createAggregate(id: String): ActorRef = {
context.actorOf(SamplePersistentActor.props(id))
}
<<<
The persistent aggregates have the following:
>>>
def initial: Receive = {
case SamplePersistentActor.GetContent => {
val info = SamplePersistentActor.Info(id, model.content, model.count)
// sender() ! info
log.info("GetContent: {}", info)
}
case SamplePersistentActor.Initialize(content) => {
if (content != "")
persist(SamplePersistentActor.AggregateAdded(content)) { evt =>
model = model.updated(evt)
context.become(created)
log.info("New aggregate saved: {}",
SamplePersistentActor.Info(id, model.content, model.count))
}
}
}
def created: Receive = {
case SamplePersistentActor.GetContent => {
val info = SamplePersistentActor.Info(id, model.content, model.count)
// sender() ! info
log.info("GetContent: {}", info)
}
case SamplePersistentActor.ChangeContent(newContent) =>
persist(SamplePersistentActor.ContentChanged(newContent)) { evt =>
model = model.updated(evt)
val info = SamplePersistentActor.Info(id, model.content, model.count)
log.info("Aggregate changed: {}", info)
}
}
<<<
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
