@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."))
}
}