I'm trying to write two ScalaTest tests to compare two approaches to
driving a Flow that sends out emails. In one approach the flow is sourced
using eventsByTag from akka-persistence-query. In the other, I uses
SourceQueue.offer after receiving a message via AtLeastOnceDelivery. The
test ends with a check of mock JavaMail mailbox size.
eventually {
mailBoxSize("ema...@foo.com") should be(expectedEmails)
}
eventually {
mailBoxSize("ema...@foo.com") should be(expectedEmails)
}
The test that uses eventsByTag to drive the flow succeeds with 10000
events. The one that uses SourceQueue.offer gets stuck at 11. In the flow
below, the println appears 12 times but the log("end-of-lookup") only
appears 11 times.
def flowLookupEmails()(implicit lookup: EmailsByDefinition, ec:
ExecutionContext): Flow[AlarmOccurrence, AlarmOccurrenceWithRecipients,
NotUsed] = {
val parallelism = 5
val result = Flow[AlarmOccurrence].mapAsync(parallelism) { occurrence =>
val emailsF: Future[List[String]] =
lookup.recipientsForDefinition(occurrence.definitionId)
val lookupResult: Future[AlarmOccurrenceWithRecipients] = emailsF.map
{ emails =>
println(s"##### Found emails $emails")
AlarmOccurrenceWithRecipients(occurrence, emails)
}
lookupResult
}
.log("end-of-lookup").withAttributes(Attributes.logLevels(onElement =
Logging.InfoLevel))
.named("lookup-emails")
result
}
When I turn on DEBUG logging, I get a lot of noise, but do see
[DEBUG] [11/09/2016 08:58:02.827]
[OccurrencesIntegrationSpec-akka.actor.default-dispatcher-12]
[akka://OccurrencesIntegrationSpec/user/StreamSupervisor-3/flow-25-0-unknown-operation]
stopped
[ERROR] [11/09/2016 08:59:02.455]
[OccurrencesIntegrationSpec-akka.actor.default-dispatcher-21]
[akka.stream.Log(akka://OccurrencesIntegrationSpec/user/StreamSupervisor-0)]
[end-of-lookup] Upstream failed. (akka.stream.AbruptTerminationException)
Any ideas? What are some good techniques to find the cause of the
AbruptTerminationException, or name the StreamSupervisors so I know which
stream they are associated with?
Thanks.
--
>>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
