Ufdjrw

I have an actor that invokes a stream. At runtime this works as intended, but when tested the stream is not invoked.

The Actor (abbreviated)

class PaymentProcessorActor(repo: PaymentRepo, accountCache: AccountCache, config: AppConfig) extends Actor {

 implicit private val materializer: ActorMaterializer = ActorMaterializer()
 implicit private val network: Network = config.network
 private implicit val ec: ExecutionContextExecutor = context.dispatcher

 val paymentSink: Sink[(Seq[Payment], Account), NotUsed] =
 Flow[(Seq[Payment], Account)].map case (ps, account) =>
 println("inside flow")
 // ... block of type Future[(TransactionResponse, Seq[Payment], Account)] here
 
 .mapAsync(parallelism = config.accounts.size)(_.map 
 case ((_: TransactionApproved, ps), account) =>
 // handle approval

 case ((x: TransactionRejected, ps), account) =>
 // handle rejection
 )
 .to(Sink.ignore)

 override def receive: Receive = state(nextKnownPaymentDate = None)

 private def state(nextKnownPaymentDate: Option[ZonedDateTime]): Receive =
 processPayments(nextKnownPaymentDate) orElse
 updateNextPaymentTime orElse
 confirmPayments orElse
 rejectPayments orElse
 rejectTransaction orElse
 retryPayments orElse
 updateAccount orElse
 registerAccount


 // If there are payments due, find and pay them
 def processPayments(nextKnownPaymentDate: Option[ZonedDateTime]): PartialFunction[Any, Unit] = 
 case ProcessPayments if nextKnownPaymentDate.exists(_.isBefore(ZonedDateTime.now())) =>
 val readyAccounts = accountCache.readyCount
 if (readyAccounts > 0) 
 val payments = repo.due(readyAccounts * 100)
 if (payments.isEmpty) 
 logger.debug("No more payments due.")
 context.become(state(repo.earliestTimeDue))
 else 
 val submittingPaymentsWithAccounts: Seq[(Seq[Payment], Account)] =
 payments.grouped(100).flatMap(ps => accountCache.borrowAccount.map(ps -> _)).toSeq
 val submittingPayments: Seq[Payment] = submittingPaymentsWithAccounts.flatMap(_._1)
 repo.submit(submittingPayments.flatMap(_.id), ZonedDateTime.now)

 Source.fromIterator(() => submittingPaymentsWithAccounts.iterator).to(paymentSink).run()
 println("post source run")
 
 
 

The spec. (sampleOf just creates a random instance and is not pertinent to the problem).

 "the payment sink" should 
 "submit to the network" in 
 val (network, conf, repo, cache) = setup
 val account = sampleOf(genAccount)
 val payments = sampleOf(Gen.listOfN(3, genPayment))
 when(repo.earliestTimeDue).thenReturn(Some(ZonedDateTime.now()))
 when(repo.due(100)).thenReturn(payments)

 val actor = system.actorOf(Props(new PaymentProcessorActor(repo, cache, conf)))

 // these two calls set up the actor state so that payments will be processed
 actor ! UpdateNextPaymentTime
 actor ! UpdateAccount(account)

 // this invokes the stream under test
 actor ! ProcessPayments

 eventually(timeout(5 seconds)) 
 assert(network.posted.size == 1)
 
 
 

 private def setup: (StubNetwork, AppConfig, PaymentRepo, AccountCache) = 
 val n = StubNetwork()
 val conf = new AppConfig 
 val network: Network = n
 val accounts: Map[String, KeyPair] = Map.empty
 
 val repo = mock[PaymentRepo]
 (n, conf, repo, new AccountCache)
 

At runtime, I see the stdout messages:

post source run
inside flow

But during test I only see

post source run

With debugging, I see that all values are correct and the source .run is called. But somehow it does not run.

In the line .mapAsync(parallelism = config.accounts.size), the value was zero, which is an error condition. The Flow never initialised. This failure does not propagate to the main thread.

Additionally, I had turned off Akka logging for tests in the configuration, so this failure was not logged.