Actor does not update its internal state

scala

(Onyancha Chrispinus) #1

We have been using Akka in our major transaction system. We are facing a small problem whereby the stateData does not update to its new value.
This is our scenario.

We have actors A,B, and C.

Actor A has an initial balance of 1,000, Actor B has an initial balance of 1,000 and
Actor C has an initial balance of $ 1,000.

Actor A sends 50 to Actor B. At this point the current logs read Actor A = 950 and Actor B = 1,050 When we check the balance status of both, Actor A reads 950 but Actor B still reads $1,000

At this point, when A sends again 100 to B. At this point the current logs read Actor A = 850 and Actor B = ,1150 When we check the balance status of both, Actor A reads 850 but Actor B still reads $1,000

When NOW Actor C sends 300 to Actor B, At this point the current logs read Actor C = 700 and Actor B now reads = 1,300. When we check the balance status of both, Actor C reads 700 but Actor B still reads $1,000

We have lost all the transactions that happened with Actor A at this point to Actor B. Can some one guide please…


(Rob Crawford) #2

How are you checking the balance? Is this all on the same node of a cluster, or not a clustered environment at all? How can the logged values be different than what your “check” gives? What does the logging say for B after C sends 300?

Without example code, it’s hard to say what could be going on here.


(Onyancha Chrispinus) #3

@Crawford Thank you very much for the feedback. I would like to notify you that this is not in the Cluster, its just one node machine.



class TransactionActor @Inject()(@Assisted phoneNo: String, @Assisted systemActor: ActorRef, mdb: AccountDao, cdb: CacheApi, userFactory: ClinicPesaActor.Factory, bus: LookupBusImpl)(implicit mat: Materializer, ec: ExecutionContext, configuration: Configuration)
  extends Actor with InjectedActorSupport with Processor with Processor_PTW with ActorLogging with Stash {

  //  import context._

  private val marker = LogMarker(name = self.path.name)

  private val ass: TransactionAss = new TransactionAss(mdb)

  private val accountState = AccountState()

  startWith(Idle, Uninitialized)


  when(Idle) {

    case Event((pw: String, balance: T_Balance, state: Boolean), _) =>
      log.info(marker, "EVENT: INIT NEW BALANCE")
      val newData = accountState.copy(verification = if (pw == msg_passcode) None else Some(pw) balance = Some(balance), isOpen = state)
      unstashAll()
      goto(ActiveRun) using newData

    case Event((UN_PAUSE, balance: T_Balance), accountState: AccountState) =>
      log.info(marker, "EVENT: UPDATE NEW BALANCE")
      val result_bal = balance match {
        case bal: T_Balance =>   Some(bal)
        case _ =>  accountState balance
      }
      val newData = accountState.copy(balance = result_bal)
      log.info(marker, s"EVENT: UPDATED NEW BALANCE =  $newData ")
      unstashAll()
      goto(ActiveRun) using newData

    case _ =>
      log.info(marker, "EVENT: TEMP STASH")
      stash()
      stay

  }


  when(ActiveRun) {
    /**
      * Listen to the all external commands by the user $phoneNo.
      */
    case Event(PhoneNumberTask(replyTo, task), accountState: AccountState) =>
      log.info(marker, s"EVENT: CLIENT COMMAND")
      this.processRequestMOM((replyTo, task, accountState))
      stay 

    /**
      * Listen to an event to temporarily pose transactions
      */
    case Event(PAUSE, accountState) =>
      log.info(marker, "EVENT: ACTIVATE PAUSE")
      goto(Idle) 

    case Event(((WITHDRAW, INITIATED, transactionS: TransactionS), _: ActorRef, replyTo: ActorRef, names: String), accountState: AccountState) =>
      log.info(marker, "EVENT: INIT WITHDRAW")
      replyTo ! DoneTrans(MessageApiTrans(transactionS.accountNo, transactionS.t_id, names, s"${transactionS.currency} ${transactionS.transFees.getOrElse(0.0)}"))
      stay //using accountState


    /**
      * Listen to the internal  Process WITHDRAW_PAYMENT and WITHDRAW_TRANSFER call of the payer
      */
    case Event(((WITHDRAW, PROCESSING, transactionS: TransactionS), _: ActorRef, replyTo: ActorRef), accountState: AccountState) =>
      log.info(marker, "EVENT:PROCESS WITHDRAW")
      self ! PAUSE
      updateWithdraw(transactionS.copy(status = PROCESSING), accountState, getSendeeActor(transactionS.paymentTo.get, accountState.context.get, systemActor, userFactory), replyTo, mdb).pipeTo(self).map(freeOurResources(_, self))
      stay

    /**
      * Listen to the internal Process DEPOSIT event completion received from Payer to Payee
      */
    case Event((DEPOSIT, transactionS: TransactionS, replyTo: ActorRef), accountState: AccountState) =>
      log.info(marker, "EVENT: DEPOSIT INTERNAL")
      self ! PAUSE
      updateDeposit(transactionS, accountState, mdb, replyTo).map(freeOurResources(_, self))
      stay 

    /**
      * Listen to the Internal message during fails and on completion  of a given task.
      */
    case Event((message@(_: Failed | _: Done), replyTo: ActorRef), _: AccountState) =>
      log.info(marker, "EVENT: ON COMPLETE")
      replyTo forward message
      stay //using accountState

  }


  whenUnhandled {

    case Event(cxt: ActorContext, accountState: AccountState) =>
      log.info(marker, "EVENT: UPDATE CONTEXT")
      stay using accountState.copy(context = Some(cxt))

    case Event(GetNames, accountState: AccountState) =>
      log.info(marker, "EVENT: GET NAMES")
      var names = receipientAccountNotFound(phoneNo)
      if (accountState.isOpen)
        names = accountState.balance.get.names
      sender ! (names, accountState.isOpen)
      stay 

    case _ =>
      log.info(marker, "EVENT: LOST")
      stay 
  }

  onTransition {
    case Idle -> ActiveRun => log.debug(marker, s"****:: ${self.path.name} TRANSITION  Idle -> ActiveRun :::  $nextStateData")
    case ActiveRun -> Idle => log.debug(marker, s"****:: ${self.path.name} TRANSITION  ActiveRun -> Idle :::  $nextStateData")
  }

 
  def processRequestMOM(valuePassed: (ActorRef, Any, AccountState)): Unit = valuePassed._2 match {
    case GetBalance =>
      log.info(marker, s"EVENT: GET BALANCE  $stateData")
      val result = valuePassed._3.balance.getOrElse(Failed(MessageApi(phoneNo, retrieve_failed)))
      valuePassed._1 ! result
      
    case value: TransactionCMD =>
      log.info(marker, s"EVENT: PROCESS ${value.transactionType} CMD")
      processInit_T(valuePassed, value, phoneNo, mdb, getSendeeActor(value.reciever, valuePassed._3.context.get, systemActor, userFactory)) pipeTo self

    case value@Confirm(_, reciever, _, _) =>
      log.info(marker, s"EVENT: CONFIRM TXT P|T CMD")
      confirmTransaction(value, getSendeeActor(reciever, valuePassed._3.context.get, systemActor, userFactory), valuePassed, mdb) pipeTo self
  }


  initialize()

  initialiseActor(phoneNo, mdb).map[Unit] {
    result: (String, T_Balance, Boolean) =>
      self ! result
      unstashAll()
      goto(ActiveRun) using accountState.copy(verification = if (result._1 != msg_passcode) None else Some(result._1), balance = Some(result._2))
  }

}


object TransactionActor {

  trait Factory {
    def apply(id: String, systemActor: ActorRef): Actor
  }

}



trait Processor_PTW {
  def ourReciever(result: Any): (Any, Boolean) = result match {
    case (outPut: String, true) => (outPut, true)
    case (otherResult, false) => (otherResult, false)
  }

  private def process(replyTo: ActorRef, transaction: TransactionS, balance: T_Balance, value: TransactionCMD, phoneNo: String, mdb: AccountDao, eventualRef: Future[ActorRef], transType: TRANSACTION_TYPE)(implicit configuration: Configuration, ec: ExecutionContext, timeout: akka.util.Timeout): Future[Any] = {
    if (balance.cashAmount >= value.cashAmount + transaction.transFees.getOrElse(0.0))
      for {
        result <- mdb.setTransaction(transaction)
        if result._2 == SUCCESS
        sendee <- eventualRef
        nameOut <- (sendee ? GetNames).map(ourReciever)
      } yield if (result._2 == SUCCESS && nameOut._2) ((transType, INITIATED, transaction), sendee, replyTo, nameOut._1) else (Failed(MessageApi(phoneNo, if (!nameOut._2) s"${nameOut._1}" else initiationFailed(phoneNo))), replyTo)
    else
      Future.successful((Failed(MessageApi(phoneNo, insufficient_balance(phoneNo, balance.currency + "  " + balance.cashAmount))), replyTo))
  }



  def processInit_T(valuePassed: (ActorRef, Any, AccountState), value: TransactionCMD, phoneNo: String, mdb: AccountDao, eventualRef: Future[ActorRef])(implicit configuration: Configuration, ec: ExecutionContext, timeout: akka.util.Timeout): Future[Any] /*(TRANSACTION_TYPE, STATUS, TransactionS)*/ = {
    val balance = valuePassed._3.balance.get
    val transaction = TransactionS(phoneNo, value.cashAmount, balance.cashAmount, 0.0, value.transactionId, balance.currency, Some(value.reason), value.transactionType, Some(value.reciever), Some(transFees(value.transactionType, value.cashAmount)), Status.INITIATED, value.timeStamp, value.timeStamp)
    process(valuePassed._1, transaction, balance, value, phoneNo, mdb, eventualRef, WITHDRAW)
  }.recover {
    case _: Exception => (Failed(MessageApi(phoneNo, initiationFailed(phoneNo))), valuePassed._1)
  }


  def confirmTransaction(value: Confirm, eventualRef: Future[ActorRef], valuePassed: (ActorRef, Any, AccountState), mdb: AccountDao)(implicit configuration: Configuration, ec: ExecutionContext): Future[Any] = {
    if (getPassword(value.pinCode) == valuePassed._3.verification.getOrElse(msg_passcode)) {
      for {
        result <- mdb.getTransaction(value.transactionId, value.sender, value.reciever, INITIATED toString) 
        if result._2 == SUCCESS
        sendee <- eventualRef
      } yield if (result._2 == SUCCESS) ((WITHDRAW, PROCESSING, result._1), sendee, valuePassed._1) else (Failed(MessageApi(value.sender, initiationFailed(value.sender))), valuePassed._1)
    } else {
      Future.successful((Failed(MessageApi(value.sender, wrong_pin_confirmation)), valuePassed._1))
    }
  }.recover { case _: Exception =>
    (Failed(MessageApi(value.sender, trans_not_found_init(value.transactionId))), valuePassed._1)
  }


  def confirmWithdraw(value: Confirm_W, phoneNo: String, valuePassed: (ActorRef, Any, AccountState), mdb: AccountDao)(implicit configuration: Configuration, ec: ExecutionContext): Future[Any] = {
    val balance = valuePassed._3.balance.get
    if (getPassword(value.pinCode) == valuePassed._3.verification.getOrElse(msg_passcode)) {
      for {
        details <- mdb.getBankAccDetails(phoneNo)
        transactionS <- mdb.getTransaction(value.transactionId, phoneNo, details._1.accountNo, INITIATED toString)  
        if transactionS._2 == SUCCESS
        result <- mdb.updateTransaction(transactionS._1.status, transactionS._1.amount, balance cashAmount, 0.0, transactionS._1.accountNo, transactionS._1.t_id, PROCESSING.toString)
        _ <- mdb.setTransaction(transactionS._1.copy(accountNo = system_Key, paymentTo = Some(transactionS._1.accountNo), transType = T_FEES , amount = transactionS._1.transFees.get, transFees = Some(0.0), status = PROCESSING))
      } yield if (result._2 == SUCCESS)((transactionS, result), valuePassed._1) else (Failed(MessageApi(phoneNo, initiationFailed(phoneNo))), valuePassed._1)
    } else {
      Future.successful((Failed(MessageApi(phoneNo, wrong_pin_confirmation)), valuePassed._1))
    }
  }.recover { case _: Exception =>
    (Failed(MessageApi(phoneNo, trans_not_found_init(value.transactionId))), valuePassed._1)
  }


  def updateWithdraw(transactionS: TransactionS, accountState: AccountState, eventualRef: Future[ActorRef], parentActor: ActorRef, mdb: AccountDao)(implicit configuration: Configuration, ec: ExecutionContext): Future[Any] = {
    val t_balance: T_Balance = accountState.balance.get
    val balance = t_balance.copy(cashAmount = t_balance.cashAmount - (transactionS.amount + transactionS.transFees.get), id = getUUID)
    for {
      updateBal <- Account.savek(transactionS.accountNo, balance cashAmount)
      if updateBal._2 == SUCCESS // What happens when it updates the this and fails on the next???
      result <- mdb.updateTransaction(INITIATED, transactionS.amount, balance cashAmount, 0.0,  transactionS.accountNo, transactionS.t_id, transactionS.status.toString)
      if result._2 == SUCCESS
      sendee <- eventualRef
    } yield if (updateBal._2 == SUCCESS && result._2 == SUCCESS) (transactionS.copy(currentBalancePayee = balance cashAmount), sendee, parentActor, balance) else (Failed(MessageApi(transactionS.accountNo, initiationFailed(transactionS.accountNo))), parentActor)

  }.recover { case _: Exception =>
    (Failed(MessageApi(transactionS.accountNo, initiationFailed(transactionS.accountNo))), parentActor)
  }


  def updateDeposit(trans: TransactionS, accountState: AccountState, mdb: AccountDao, replyTo: ActorRef)(implicit configuration: Configuration, ec: ExecutionContext): Future[Any] = {

    val transactionS = trans.copy(status = CONFIRMED, timeStamp = DateTime.now.getMillis)
    val t_balance: T_Balance = accountState.balance.get
    System.out.println(s"WE ARE DEPOSITING old balance = ${t_balance.cashAmount}  new amount to add  ${transactionS.amount}")
    val balance = t_balance.copy(cashAmount = t_balance.cashAmount + transactionS.amount, id = getUUID)
    System.out.println(s" NOW THE BALANCE = $balance")
    for {
      updateBal <- Account.savek(transactionS.paymentTo.get, balance cashAmount)
      if updateBal._2 == SUCCESS
      result <- mdb.updateTransaction(PROCESSING, transactionS.amount, trans currentBalancePayee, balance cashAmount, transactionS.accountNo, transactionS.t_id, transactionS.status.toString)
    } yield if (updateBal._2 == SUCCESS && result._2 == SUCCESS) (Done(MessageApi(transactionS.accountNo, s"Transaction to ${transactionS.paymentTo.get} of  ${transactionS.amount} successful")), balance, replyTo) else Failed(MessageApi(transactionS.accountNo, initiationFailed(transactionS.accountNo)))
  }.recover { case _: Exception =>
    Failed(MessageApi(trans.accountNo, initiationFailed(trans.accountNo)))
  }


  def freeOurResources(result: Any, currentActor: ActorRef): Unit = result match {
    case (transactionS: TransactionS, sendee: ActorRef, replyTo, t_balance: T_Balance) =>
      System.out.println(s"WE transactionS  ***************** ${currentActor.path}  $t_balance")
      currentActor ! (UN_PAUSE, t_balance)
      sendee ! (DEPOSIT, transactionS, replyTo)
    case _: Failed =>
      currentActor ! (UN_PAUSE, null)
    case (output: Done, t_balance: T_Balance, sendee: ActorRef) =>
      System.out.println(s"WE Done  ***************** ${currentActor.path}    $t_balance")
      currentActor ! (UN_PAUSE, t_balance)
      sendee ! output
  }

  
}


trait Processor extends ClinicPesaInitializer {

 def extractBalance(value: Option[Account]): (T_Balance, Boolean) = {
    val balance = T_Balance(Currency.UNDEFINED, 0.0, "", getUUID)
    value match {
      case Some(fund: Account) => (balance.copy(currency = getCurrency(fund.contact.get), cashAmount = fund.balance, names = fund.names), true)
      case None => (balance, false)
    }
  }

  def initialiseActor(phoneNo: String, mdb: AccountDao)(implicit ec: ExecutionContext): Future[(String, T_Balance, AutoSave, Boolean)] =
    for {
      balance <- Account.find(phoneNo).map(extractBalance)
      verify <- mdb.getVer(phoneNo).map {
        case (value, SUCCESS) => value
        case _ => Messages.msg_passcode
      }
    } yield (verify, balance._1, balance._2)


  def transactionInitiated(result: (String, MESSAGE), phone:String): MessagesOpStatus = result match {
    case (_, SUCCESS) =>
      Done (MessageApi(result._1.substring(0,7), s"Transaction successfully initiated to Mobile Money Account $phone."))
    case (_, FAILURE) =>
      Failed(MessageApi(result._1, "Transaction was unsuccessfully in the initiation process."))
  }

}


(Rob Crawford) #4

I’m not an expert on the FSM approach, but shouldn’t the “stay” at the end of WITHDRAW/PROCESSING and DEPOSIT have a “using(accountState)”?


(Onyancha Chrispinus) #5

Thank you very much @Crawford. using (accountState) is only necessary when changing the dataState. It maintains the old State data.

The logs are here…


On checking balance for actor clinicPesa-user-256788393954 before transaction

[debug] a.a.ActorDSL$Extension$$anon$1 - [] now supervising Actor[akka://Transaction/system/dsl/inbox-7#1980819743]
[debug] a.a.d.Inbox$InboxActor - [] started (akka.actor.dsl.Inbox$InboxActor@1adc2649)
[debug] a.UserParentActor - [] received handled message Create(256788393954,api.GetBalance$@25d9009b) from Actor[akka://Transaction/system/dsl/inbox-7#1980819743]
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$e])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: GET BALANCE  AccountState(true,Some(T_Balance(UGX,1998000,SHARON ANYANGO,048ebb1ba0004e1e85da7f4cb220b4b6)),None,Some( 256788393954 and amount of 0.0 and status INITIATED),Some(OTJkZDBiNjVkNzRjNmNiMDdhNzljMWVhMmFhODAyMTY2OWNiMmFlZmIyYWU3ZmFlNzIwNTdmOWQ0YjhhMmRjNg==),Some(akka.actor.ActorCell@7de2b3ae))

On checking balance for actor clinicPesa-user-10005 before transaction

[debug] a.a.ActorDSL$Extension$$anon$1 - [] now supervising Actor[akka://Transaction/system/dsl/inbox-2#306635348]
[debug] a.UserParentActor - [] received handled message Create(10005,api.GetBalance$@3af82cb0) from Actor[akka://Transaction/system/dsl/inbox-2#306635348]
[debug] a.a.d.Inbox$InboxActor - [] started (akka.actor.dsl.Inbox$InboxActor@44a78b05)
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$a])
[info] a.ClinicPesaActor - [clinicPesa-user-10005] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-10005] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-10005] EVENT: GET BALANCE  AccountState(true,Some(T_Balance(UGX,2006000,KWAME CLINIC,e8f060e479764c60a6abce2132bb1ac7)),None,Some( 10005 and amount of 0.0 and status INITIATED),Some(YzgzM2JiYWZmNjlhZDIzNGYzN2Y4YmIxNGZhNzNmMzA0NzhiMDQ4Yjg4ZjY2MDQ2MzFhODg2ZDYwMGZiNzJmOA==),Some(akka.actor.ActorCell@5166b9f4))



At that moment the persisted database value is:

id |   phone_no    |        names        | balance | tomb_storm |    contact    |         created_at         | deleted_at
----+---------------+---------------------+---------+------------+---------------+----------------------------+------------
67 | 256788393954  | SHARON ANYANGO      | 1998000 | f          | 256788393954  | 2017-11-11 13:54:17.568+03 |
54 | 10005         | KWAME CLINIC        | 2006000 | f          | 256752818012  | 2017-09-20 14:32:53.721+03 |

On calling transfer from actor clinicPesa-user-256788393954 to actor clinicPesa-user10005 of  amount 1,000.

[debug] a.a.ActorDSL$Extension$$anon$1 - [] now supervising Actor[akka://Transaction/system/dsl/inbox-6#-2059972309]
[debug] a.UserParentActor - [] received handled message Create(256788393954,TransactionCMD(10005,Malaria,1000,eef5ad8c2ef34649b2263e7b14002bae,1528207141315,PAYMENT)) from Actor[akka://Transaction/system/dsl/inbox-6#-2059972309]
[debug] a.a.d.Inbox$InboxActor - [] started (akka.actor.dsl.Inbox$InboxActor@78ff89d1)
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$c])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: PROCESS PAYMENT CMD
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$d])
[info] a.ClinicPesaActor - [clinicPesa-user10005] EVENT: GET NAMES
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: INIT WITHDRAW


on Init payment
[info] c.g.m.a.d.p.c.PostgreSQLConnectionHandler - [] Connection disconnected - localhost/127.0.0.1:5432
[debug] a.a.ActorDSL$Extension$$anon$1 - [] now supervising Actor[akka://Transaction/system/dsl/inbox-4#-1841051368]
[debug] a.UserParentActor - [] received handled message Create(256788393954,TransactionCMD(10005,Malaria,1000,37f02628a4414ae9806f898f984826b8,1528207413641,PAYMENT)) from Actor[akka://Transaction/system/dsl/inbox-4#-1841051368]
[debug] a.a.d.Inbox$InboxActor - [] started (akka.actor.dsl.Inbox$InboxActor@6f5ccd89)
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$b])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: PROCESS PAYMENT CMD
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$c])
[info] a.ClinicPesaActor - [clinicPesa-user10005] EVENT: GET NAMES
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: INIT WITHDRAW


On Confirming transaction here

[debug] a.UserParentActor - [] received handled message Create(256788393954,Confirm(256788393954,10005,37f02628a4414ae9806f898f984826b8,1111)) from Actor[akka://Transaction/system/dsl/inbox-5#-1530543343]
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$d])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: CONFIRM TXT P|T CMD
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$e])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT:PROCESS WITHDRAW
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$f])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: ACTIVATE PAUSE
[debug] a.ClinicPesaActor - [clinicPesa-user-256788393954] ****:: clinicPesa-user-256788393954 TRANSITION  ActiveRun -> Idle :::  AccountState(true,Some(T_Balance(UGX,1998000,SHARON ANYANGO,b4452c71de43498f907cdf67d2e78a3b)),None,Some( 256788393954 and amount of 0.0 and status INITIATED),Some(OTJkZDBiNjVkNzRjNmNiMDdhNzljMWVhMmFhODAyMTY2OWNiMmFlZmIyYWU3ZmFlNzIwNTdmOWQ0YjhhMmRjNg==),Some(akka.actor.ActorCell@5166b9f4))
check WrappedArray(256788393954, 37f02628a4414ae9806f898f984826b8, PROCESSING)  INITIATED   1000.0
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: TEMP STASH
WE transactionS  ***************** akka://application/user/userParentActor/clinicPesa-user-256788393954  T_Balance(UGX,1997000.0,SHARON ANYANGO,f1eee2ba9ea54038a4f91beb86018491)
WE ARE DEPOSITING old balance = 2006000  new amount to add  1000.0
 NOW THE BALANCE = T_Balance(UGX,2007000.0,KWAME CLINIC,2446202f02344c65b6d7cce163500f6c)
[info] a.ClinicPesaActor - [clinicPesa-user10005] EVENT: DEPOSIT INTERNAL
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATE NEW BALANCE
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATED NEW BALANCE =  AccountState(true,Some(T_Balance(UGX,1997000.0,SHARON ANYANGO,f1eee2ba9ea54038a4f91beb86018491)),None,Some( 256788393954 and amount of 0.0 and status INITIATED),Some(OTJkZDBiNjVkNzRjNmNiMDdhNzljMWVhMmFhODAyMTY2OWNiMmFlZmIyYWU3ZmFlNzIwNTdmOWQ0YjhhMmRjNg==),Some(akka.actor.ActorCell@5166b9f4))
[debug] a.ClinicPesaActor - [clinicPesa-user-256788393954] ****:: clinicPesa-user-256788393954 TRANSITION  Idle -> ActiveRun :::  AccountState(true,Some(T_Balance(UGX,1997000.0,SHARON ANYANGO,f1eee2ba9ea54038a4f91beb86018491)),None,Some( 256788393954 and amount of 0.0 and status INITIATED),Some(OTJkZDBiNjVkNzRjNmNiMDdhNzljMWVhMmFhODAyMTY2OWNiMmFlZmIyYWU3ZmFlNzIwNTdmOWQ0YjhhMmRjNg==),Some(akka.actor.ActorCell@5166b9f4))
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: LOST
[info] a.ClinicPesaActor - [clinicPesa-user10005] EVENT: ACTIVATE PAUSE
[debug] a.ClinicPesaActor - [clinicPesa-user10005] ****:: clinicPesa-user10005 TRANSITION  ActiveRun -> Idle :::  AccountState(true,Some(T_Balance(UGX,2006000,KWAME CLINIC,080949d50b0c42a4985db91cd80f9238)),None,Some( 10005 and amount of 0.0 and status INITIATED),Some(YzgzM2JiYWZmNjlhZDIzNGYzN2Y4YmIxNGZhNzNmMzA0NzhiMDQ4Yjg4ZjY2MDQ2MzFhODg2ZDYwMGZiNzJmOA==),None)
check WrappedArray(256788393954, 37f02628a4414ae9806f898f984826b8, CONFIRMED)  PROCESSING   1000.0
WE Done  ***************** akka://application/user/userParentActor/clinicPesa-user10005    T_Balance(UGX,2007000.0,KWAME CLINIC,2446202f02344c65b6d7cce163500f6c)
[info] a.ClinicPesaActor - [clinicPesa-user10005] EVENT: UPDATE NEW BALANCE
[info] a.ClinicPesaActor - [clinicPesa-user10005] EVENT: UPDATED NEW BALANCE =  AccountState(true,Some(T_Balance(UGX,2007000.0,KWAME CLINIC,2446202f02344c65b6d7cce163500f6c)),None,Some( 10005 and amount of 0.0 and status INITIATED),Some(YzgzM2JiYWZmNjlhZDIzNGYzN2Y4YmIxNGZhNzNmMzA0NzhiMDQ4Yjg4ZjY2MDQ2MzFhODg2ZDYwMGZiNzJmOA==),None)
[debug] a.ClinicPesaActor - [clinicPesa-user10005] ****:: clinicPesa-user10005 TRANSITION  Idle -> ActiveRun :::  AccountState(true,Some(T_Balance(UGX,2007000.0,KWAME CLINIC,2446202f02344c65b6d7cce163500f6c)),None,Some( 10005 and amount of 0.0 and status INITIATED),Some(YzgzM2JiYWZmNjlhZDIzNGYzN2Y4YmIxNGZhNzNmMzA0NzhiMDQ4Yjg4ZjY2MDQ2MzFhODg2ZDYwMGZiNzJmOA==),None)
[info] c.g.m.a.d.p.c.PostgreSQLConnectionHandler - [] Connection disconnected - localhost/127.0.0.1:5432

id |   phone_no    |        names        | balance | tomb_storm |    contact    |         created_at         | deleted_at
----+---------------+---------------------+---------+------------+---------------+----------------------------+------------
43 | 256752818011  | ONYANCHA CHRISPINUS | 1986000 | f          | 256752818011  | 2017-09-20 12:04:33.03+03  |
54 | 10005         | KWAME CLINIC        | 2007000 | f          | 256752818012  | 2017-09-20 14:32:53.721+03 |

On checking balance for actor clinicPesa-user-256788393954 after transaction

[debug] a.a.ActorDSL$Extension$$anon$1 - [] now supervising Actor[akka://Transaction/system/dsl/inbox-7#1272937586]
[debug] a.a.d.Inbox$InboxActor - [] started (akka.actor.dsl.Inbox$InboxActor@75010c52)
[debug] a.UserParentActor - [] received handled message Create(256788393954,api.GetBalance$@3af82cb0) from Actor[akka://Transaction/system/dsl/inbox-7#1272937586]
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$h])
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-256788393954] EVENT: GET BALANCE  AccountState(true,Some(T_Balance(UGX,1997000.0,SHARON ANYANGO,f1eee2ba9ea54038a4f91beb86018491)),None,Some( 256788393954 and amount of 0.0 and status INITIATED),Some(OTJkZDBiNjVkNzRjNmNiMDdhNzljMWVhMmFhODAyMTY2OWNiMmFlZmIyYWU3ZmFlNzIwNTdmOWQ0YjhhMmRjNg==),Some(akka.actor.ActorCell@5166b9f4))

On checking balance for actor clinicPesa-user-10005 after transaction

[debug] a.a.d.Inbox$InboxActor - [] stopped
[debug] a.a.d.Inbox$InboxActor - [] stopped
[debug] a.a.ActorDSL$Extension$$anon$1 - [] now supervising Actor[akka://Transaction/system/dsl/inbox-8#1893744115]
[debug] a.a.d.Inbox$InboxActor - [] started (akka.actor.dsl.Inbox$InboxActor@6f99df5e)
[debug] a.UserParentActor - [] received handled message Create(10005,api.GetBalance$@3af82cb0) from Actor[akka://Transaction/system/dsl/inbox-8#1893744115]
[debug] a.ClinicPesaActor - [] received AutoReceiveMessage Envelope(Identify(None),Actor[akka://application/temp/$i])
[info] a.ClinicPesaActor - [clinicPesa-user-10005] EVENT: UPDATE CONTEXT
[info] a.ClinicPesaActor - [clinicPesa-user-10005] EVENT: CLIENT COMMAND
[info] a.ClinicPesaActor - [clinicPesa-user-10005] EVENT: GET BALANCE  AccountState(true,Some(T_Balance(UGX,2006000,KWAME CLINIC,e8f060e479764c60a6abce2132bb1ac7)),None,Some( 10005 and amount of 0.0 and status INITIATED),Some(YzgzM2JiYWZmNjlhZDIzNGYzN2Y4YmIxNGZhNzNmMzA0NzhiMDQ4Yjg4ZjY2MDQ2MzFhODg2ZDYwMGZiNzJmOA==),Some(akka.actor.ActorCell@5166b9f4))

At the end, clinicPesa-user-10005 after transaction still reads 2006000 and e8f060e479764c60a6abce2132bb1ac7 instead of 2007000 and 2446202f02344c65b6d7cce163500f6c.


(Patrik Nordwall) #6

I suspect that you access the actor state or context from the futures (or their callbacks). When mixing actors and futures the result of the result of the futures mus be sent to the actor as messages. See pipeTo pattern in documentation.


(Onyancha Chrispinus) #7

Thank you @patriknw. Let me go through the futures and make sure all result feedback is returned to the Actor through the pipeTo pattern.