-
Notifications
You must be signed in to change notification settings - Fork 20
Worked Example : Bank Account
To illustrate the main capabilities of event sourcing backed entities we have the following superficial example of a bank account management system. There is a Blazor front end here that you can test these functions with.
There are four events in the initial minimum viable product of our example system:
- Account Opened occurs when a new account is created
- Beneficiary Set occurs when the person who owns the account for legal purposes is set
- Money deposited occurs when money is paid into the account
- Money withdrawn occurs when money is taken out of the account
The account is uniquely identified by an account number - any unique string will do for this.
- An account can only be created once - therefore if an event stream already exists for the account, do not create a new one or append an account opened event.
- For all of the other events an account must already exist and have been opened before they can be applied.
- For money withdrawn we need to make sure there are sufficient funds in the account and this must be protected from concurrent withdrawal events.
Each account is identified by being in the domain Bank and the entity type Account so only the account number (being the entity instance unique key) is provided by the Azure serverless function. The function is allowed to specify an opening balance and a beneficiary name to illustrate how one command can write more than one event.
[FunctionName("OpenAccount")]
public static async Task<HttpResponseMessage> OpenAccountRun(
[HttpTrigger(AuthorizationLevel.Function, "POST", Route = "OpenAccount/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[EventStream("Bank", "Account", "{accountnumber}")] EventStream bankAccountEvents)
{
if (await bankAccountEvents.Exists())
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} already exists");
}
else
{
// Get request body
AccountOpeningData data = await req.Content.ReadAsAsync<AccountOpeningData>();
// Append a "created" event
DateTime dateCreated = DateTime.UtcNow;
Account.Events.Opened evtOpened = new Account.Events.Opened() { LoggedOpeningDate = dateCreated };
if (!string.IsNullOrWhiteSpace(data.Commentary))
{
evtOpened.Commentary = data.Commentary;
}
await bankAccountEvents.AppendEvent(evtOpened);
// If there is an initial deposit in the account opening data, append a "deposit" event
if (data.OpeningBalance.HasValue)
{
Account.Events.MoneyDeposited evtInitialDeposit = new Account.Events.MoneyDeposited()
{
AmountDeposited = data.OpeningBalance.Value,
LoggedDepositDate = dateCreated,
Commentary = "Opening deposit"
};
await bankAccountEvents.AppendEvent(evtInitialDeposit);
}
// If there is a beneficiary in the account opening data append a "beneficiary set" event
if (!string.IsNullOrEmpty(data.ClientName))
{
Account.Events.BeneficiarySet evtBeneficiary = new Account.Events.BeneficiarySet()
{ BeneficiaryName = data.ClientName };
await bankAccountEvents.AppendEvent(evtBeneficiary);
}
return req.CreateResponse(System.Net.HttpStatusCode.Created, $"Account {accountnumber} created");
}
}Note that the Exists() call is a non blocking call so it is possible that another function might trigger at the same time as this one and give rise to a concurrency issue. If that is a concern there is an additional optional parameter that you can set in the AppendEvent call named streamConstraint. Setting this to MustBeNew will cause a run time error to occur if an attempt is made to write to an event stream which some other process has already created.
try
{
await bankAccountEvents.AppendEvent(evtOpened,
streamConstraint: EventStreamExistenceConstraint.MustBeNew
);
}
catch (EventStreamWriteException exWrite)
{
return req.CreateResponse(System.Net.HttpStatusCode.Conflict,
$"Account {accountnumber} had a conflict error on creation {exWrite.Message }");
}To get the balance of the account we need to run a projection over the bank account event stream which handles the money deposited event and the money withdrawn event to give the balance as at a given point.
public class Balance
: ProjectionBase,
IHandleEventType<MoneyDeposited>,
IHandleEventType<MoneyWithdrawn >
{
private decimal currentBalance;
/// <summary>
/// The current balance after the projection has run over a bank account event stream
/// </summary>
public decimal CurrentBalance
{
get
{
return currentBalance;
}
}
public void HandleEventInstance(MoneyDeposited eventInstance)
{
if (null != eventInstance )
{
currentBalance += eventInstance.AmountDeposited;
}
}
public void HandleEventInstance(MoneyWithdrawn eventInstance)
{
if (null != eventInstance )
{
currentBalance -= eventInstance.AmountWithdrawn;
}
}
}Then this needs to be run from the Azure function in order to get the balance for the requested account number.
[FunctionName("GetBalance")]
public static async Task<HttpResponseMessage> GetBalanceRun(
[HttpTrigger(AuthorizationLevel.Function, "GET", Route = "GetBalance/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[Projection("Bank", "Account", "{accountnumber}", nameof(Balance))] Projection prjBankAccountBalance)
{
string result = $"No balance found for account {accountnumber}";
if (null != prjBankAccountBalance)
{
if (await prjBankAccountBalance.Exists())
{
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>();
if (null != projectedBalance)
{
result = $"Balance for account {accountnumber} is ${projectedBalance.CurrentBalance} (As at record {projectedBalance.CurrentSequenceNumber}) ";
}
}
else
{
result = $"Account {accountnumber} is not yet created - cannot retrieve a balance for it";
}
}
return req.CreateResponse(System.Net.HttpStatusCode.OK, result);
}One huge advantage of event sourcing is that is easy to get the state of an entity as at a given point in time by only running the projection over the event stream up until that point in time. To do this for the Get Balance function you would change it thus:
if (await prjBankAccountBalance.Exists())
{
// Get request body
Nullable<DateTime> asOfDate = null;
if (null != req.Content)
{
dynamic data = await req.Content.ReadAsAsync<object>();
if (null != data)
{
asOfDate = data.AsOfDate;
}
}
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>(asOfDate );
if (null != projectedBalance)
{
result = $"Balance for account {accountnumber} is ${projectedBalance.CurrentBalance} (As at record {projectedBalance.CurrentSequenceNumber}) ";
}
}Depositing money is the simplest of all operations - we simply need to be sure that the account exists.
[FunctionName("DepositMoney")]
public static async Task<HttpResponseMessage> DepositMoneyRun(
[HttpTrigger(AuthorizationLevel.Function, "POST", Route = "DepositMoney/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[EventStream("Bank", "Account", "{accountnumber}")] EventStream bankAccountEvents)
{
if (!await bankAccountEvents.Exists())
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} does not exist");
}
else
{
// get the request body...
MoneyDepositData data = await req.Content.ReadAsAsync<MoneyDepositData>();
// create a deposited event
DateTime dateDeposited = DateTime.UtcNow;
Account.Events.MoneyDeposited evDeposited = new Account.Events.MoneyDeposited()
{
LoggedDepositDate = dateDeposited,
AmountDeposited = data.DepositAmount,
Commentary = data.Commentary,
Source = data.Source
};
await bankAccountEvents.AppendEvent(evDeposited);
return req.CreateResponse(System.Net.HttpStatusCode.OK, $"{data.DepositAmount} deposited to account {accountnumber} ");
}
}Withdrawing money requires running the balance projection to make sure that the account has the funds available to withdraw and only if it does, post the withdrawal event. When posting the withdrawal event we pass in the sequence number returned by the projection and if any other events have been written to the event stream since then an error is thrown and our withdrawal event is not appended.
[FunctionName("WithdrawMoney")]
public static async Task<HttpResponseMessage> WithdrawMoneyRun(
[HttpTrigger(AuthorizationLevel.Function, "POST", Route = "WithdrawMoney/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[EventStream("Bank", "Account", "{accountnumber}")] EventStream bankAccountEvents,
[Projection("Bank", "Account", "{accountnumber}", nameof(Balance))] Projection prjBankAccountBalance)
{
if (!await bankAccountEvents.Exists())
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} does not exist");
}
else
{
// get the request body...
MoneyWithdrawnData data = await req.Content.ReadAsAsync<MoneyWithdrawnData>();
// get the current account balance
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>();
if (null != projectedBalance)
{
if (projectedBalance.CurrentBalance >= data.AmountWithdrawn)
{
// attempt the withdrawal
DateTime dateWithdrawn = DateTime.UtcNow;
Account.Events.MoneyWithdrawn evWithdrawn = new Account.Events.MoneyWithdrawn()
{
LoggedWithdrawalDate = dateWithdrawn,
AmountWithdrawn = data.AmountWithdrawn ,
Commentary = data.Commentary
};
try
{
await bankAccountEvents.AppendEvent(evWithdrawn, projectedBalance.CurrentSequenceNumber);
}
catch (EventSourcingOnAzureFunctions.Common.EventSourcing.Exceptions.EventStreamWriteException exWrite )
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Failed to write withdrawal event {exWrite.Message}");
}
return req.CreateResponse(System.Net.HttpStatusCode.OK, $"{data.AmountWithdrawn } withdrawn from account {accountnumber} ");
}
else
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} does not have sufficent funds for the withdrawal of {data.AmountWithdrawn} (Current balance: {projectedBalance.CurrentBalance} )");
}
}
else
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Unable to get current balance for account {accountnumber}");
}
}
}You could implement a retry process to run the projection and if there is still sufficient balance then attempt to post the event a second time but you should always be prepared for the command to fail if this cannot be done.
In a real system, new requirements and changes will be required throughout the project's entire existence. To illustrate how this would happen in an event sourced world I am adding a change to allow for an overdraft facility to be added to an existing account to allow the account to have a balance below zero.
[EventName("Overdraft Limit Set")]
public class OverdraftLimitSet
{
/// <summary>
/// The maximum amount the account can be overdrawn
/// </summary>
public decimal OverdraftLimit { get; set; }
/// <summary>
/// Additional notes on the overdraft being set
/// </summary>
public string Commentary { get; set; }
}This event is appended to the bank account's event stream to indicate that a new overdraft limit is in effect. Any attempt to make a withdrawal after this point needs to take the overdraft limit into effect. To remove the overdraft limit we post a new overdraft limit set event with a limit of zero.
Because the overdraft limit can be changed over time we need a new projection which gets the current overdraft limit in effect. The only event this new projection cares about is the overdraft limit set event and it just updates the current limit when it encounters this event.
/// <summary>
/// The overdraft limit in force for this account
/// </summary>
public class OverdraftLimit
: ProjectionBase,
IHandleEventType<OverdraftLimitSet>
{
private decimal currentOverdraft;
/// <summary>
/// The overdraft limit found for a bank account event stream
/// </summary>
public decimal CurrentOverdraftLimit
{
get
{
return currentOverdraft;
}
}
public void HandleEventInstance(OverdraftLimitSet eventInstance)
{
if (null != eventInstance)
{
currentOverdraft = eventInstance.OverdraftLimit;
}
}
}A new command is added to the system which sets the overdraft limits. The constraints for this command are that the account must already exists and that the existing balance must not be below any new overdraft limit set - i.e. you cannot reduce the overdraft facility if the account if overdrawn by more than the new facility. (This is to illustrate a consistency issue and may not be a real world business rule)
/// <summary>
/// Set a new overdraft limit for the account
/// </summary>
[FunctionName("SetOverdraftLimit")]
public static async Task<HttpResponseMessage> SetOverdraftLimitRun(
[HttpTrigger(AuthorizationLevel.Function, "POST", Route = @"SetOverdraftLimit/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[EventStream("Bank", "Account", "{accountnumber}")] EventStream bankAccountEvents,
[Projection("Bank", "Account", "{accountnumber}", nameof(Balance))] Projection prjBankAccountBalance)
{
if (!await bankAccountEvents.Exists())
{
// You cannot set an overdraft if the account does not exist
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} does not exist");
}
else
{
// get the request body...
OverdraftSetData data = await req.Content.ReadAsAsync<OverdraftSetData>();
// get the current account balance
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>();
if (null != projectedBalance)
{
if (projectedBalance.CurrentBalance >= (0 - data.NewOverdraftLimit) )
{
// attempt to set the new overdraft limit
Account.Events.OverdraftLimitSet evOverdraftSet = new Account.Events.OverdraftLimitSet()
{
OverdraftLimit = data.NewOverdraftLimit ,
Commentary = data.Commentary
};
try
{
await bankAccountEvents.AppendEvent(evOverdraftSet, projectedBalance.CurrentSequenceNumber);
}
catch (EventSourcingOnAzureFunctions.Common.EventSourcing.Exceptions.EventStreamWriteException exWrite)
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Failed to write overdraft limit event {exWrite.Message}");
}
return req.CreateResponse(System.Net.HttpStatusCode.OK, $"{data.NewOverdraftLimit } set as the new overdraft limit for account {accountnumber} ");
}
else
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} has an outstanding balance greater than the new limit {data.NewOverdraftLimit} (Current balance: {projectedBalance.CurrentBalance} )");
}
}
else
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Unable to get current balance for account {accountnumber}");
}
}
}The withdraw money command needs to know about the overdraft in order to include it in the decision as to whether or not a withdrawal is allowed. You can either modify the Balance projection also to return the overdraft limit or use the new Overdraft Limit projection in tandem with the balance projection. (_In the latter case you need to make sure they are operating off the same Current sequence number to prevent the possibility of a concurrency issue)
[FunctionName("WithdrawMoney")]
public static async Task<HttpResponseMessage> WithdrawMoneyRun(
[HttpTrigger(AuthorizationLevel.Function, "POST", Route = @"WithdrawMoney/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[EventStream("Bank", "Account", "{accountnumber}")] EventStream bankAccountEvents,
[Projection("Bank", "Account", "{accountnumber}", nameof(Balance))] Projection prjBankAccountBalance,
[Projection("Bank", "Account", "{accountnumber}", nameof(OverdraftLimit))] Projection prjBankAccountOverdraft)
{
if (!await bankAccountEvents.Exists())
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden, $"Account {accountnumber} does not exist");
}
else
{
// get the request body...
MoneyWithdrawnData data = await req.Content.ReadAsAsync<MoneyWithdrawnData>();
// get the current account balance
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>();
if (null != projectedBalance)
{
OverdraftLimit projectedOverdraft = await prjBankAccountOverdraft.Process<OverdraftLimit>();
decimal overdraftSet = 0.00M;
if (null != projectedOverdraft )
{
if (projectedOverdraft.CurrentSequenceNumber != projectedBalance.CurrentSequenceNumber )
{
// The two projectsions are out of synch. In a real business case we would retry them
// n times to try and get a match but here we will just throw a consistency error
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden,
$"Unable to get a matching state for the current balance and overdraft for account {accountnumber}");
}
else
{
overdraftSet = projectedOverdraft.CurrentOverdraftLimit;
}
}
if ((projectedBalance.CurrentBalance + overdraftSet) >= data.AmountWithdrawn)
{
// attempt the withdrawal
DateTime dateWithdrawn = DateTime.UtcNow;
Account.Events.MoneyWithdrawn evWithdrawn = new Account.Events.MoneyWithdrawn()
{
LoggedWithdrawalDate = dateWithdrawn,
AmountWithdrawn = data.AmountWithdrawn ,
Commentary = data.Commentary
};
try
{
await bankAccountEvents.AppendEvent(evWithdrawn, projectedBalance.CurrentSequenceNumber);
}
catch (EventSourcingOnAzureFunctions.Common.EventSourcing.Exceptions.EventStreamWriteException exWrite )
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden,
$"Failed to write withdrawal event {exWrite.Message}");
}
return req.CreateResponse(System.Net.HttpStatusCode.OK,
$"{data.AmountWithdrawn } withdrawn from account {accountnumber} (New balance: {projectedBalance.CurrentBalance - data.AmountWithdrawn}, overdraft: {overdraftSet} )");
}
else
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden,
$"Account {accountnumber} does not have sufficent funds for the withdrawal of {data.AmountWithdrawn} (Current balance: {projectedBalance.CurrentBalance}, overdraft: {overdraftSet} )");
}
}
else
{
return req.CreateResponse(System.Net.HttpStatusCode.Forbidden,
$"Unable to get current balance for account {accountnumber}");
}
}
}In our example bank we accrue interest on a daily basis but only pay it on the start of each calendar month. This means that we need two separate events fro this process: Interest Accrued...
[EventName("Interest Accrued")]
public class InterestAccrued
{
/// <summary>
/// The amount of money interest accrued for the account
/// </summary>
public decimal AmountAccrued { get; set; }
/// <summary>
/// The commentary attached to the interest
/// </summary>
public string Commentary { get; set; }
/// <summary>
/// How much was the interest rate when this accrual happened
/// </summary>
public decimal InterestRateInEffect { get; set; }
/// <summary>
/// As of when was this accrual effective
/// </summary>
public DateTime AccrualEffectiveDate { get; set; }
}and Interest Paid ..
[EventName("Interest Paid")]
public class InterestPaid
{
/// <summary>
/// The amount of money interest paid for the account
/// </summary>
/// <remarks>
/// This can be negative interest was charged
/// </remarks>
public decimal AmountAccrued { get; set; }
/// <summary>
/// The commentary attached to the interest
/// </summary>
public string Commentary { get; set; }
}In order to accrue interest we need to have the account balance at the point in time for the account - which we can get by calling the account balance projection from earlier.
We get the balance as it was at midnight even though the process itself runs after midnight - this is a business requirement which also ensures that we are not at risk of any concurrency issues.
/// <summary>
/// Calculate the accrued interest and post it to the account
/// </summary>
[FunctionName("AccrueInterest")]
public static async Task<HttpResponseMessage> AccrueInterestRun(
[HttpTrigger(AuthorizationLevel.Function, "POST", Route = @"AccrueInterest/{accountnumber}")]HttpRequestMessage req,
string accountnumber,
[EventStream("Bank", "Account", "{accountnumber}")] EventStream bankAccountEvents,
[Projection("Bank", "Account", "{accountnumber}", nameof(Balance))] Projection prjBankAccountBalance)
{
// Set the start time for how long it took to process the message
DateTime startTime = DateTime.UtcNow;
if (!await bankAccountEvents.Exists())
{
// You cannot set an overdraft if the account does not exist
return req.CreateResponse<ProjectionFunctionResponse>(System.Net.HttpStatusCode.Forbidden,
ProjectionFunctionResponse.CreateResponse(startTime,
true,
$"Account {accountnumber} does not exist",
0));
}
// get the request body...
InterestAccrualData data = await req.Content.ReadAsAsync<InterestAccrualData>();
// Get the current account balance, as at midnight
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>(DateTime.Today);
if (null != projectedBalance)
{
Account.Events.InterestAccrued evAccrued = new Account.Events.InterestAccrued()
{
Commentary = data.Commentary,
AccrualEffectiveDate = DateTime.Today // set the accrual to midnight today
};
if (projectedBalance.CurrentBalance >= 0)
{
// Using the credit rate
evAccrued.AmountAccrued = data.CreditInterestRate * projectedBalance.CurrentBalance;
}
else
{
// Use the debit rate
evAccrued.AmountAccrued = data.DebitInterestRate * projectedBalance.CurrentBalance;
}
await bankAccountEvents.AppendEvent(evAccrued);
return req.CreateResponse<ProjectionFunctionResponse>(System.Net.HttpStatusCode.OK,
ProjectionFunctionResponse.CreateResponse(startTime,
false,
$"Interest accrued for account {accountnumber} is {evAccrued.AmountAccrued}",
projectedBalance.CurrentSequenceNumber),
FunctionResponse.MEDIA_TYPE);
}
else
{
return req.CreateResponse<ProjectionFunctionResponse>(System.Net.HttpStatusCode.Forbidden,
ProjectionFunctionResponse.CreateResponse(startTime,
true,
$"Unable to get current balance for account {accountnumber} for interest accrual",
0),
FunctionResponse.MEDIA_TYPE
);
}
}This will append an interest accrued event to the event stream of the account. However we do not want to post an interest accrual for any account if it has already had an interest accrual for today. To find that out we would need to use a classification of "has had interest accrued today".
/// <summary>
/// A classification of a bank account to state whether that account has had interest
/// accrued today
/// </summary>
public class InterestAccruedToday
: ClassificationBase,
IClassifyEventType<InterestAccrued>
{
public ClassificationResponse.ClassificationResults ClassifyEventInstance(InterestAccrued eventInstance)
{
// if it happened today set it to true
if (eventInstance.AccrualEffectiveDate.Date == DateTime.Today )
{
return ClassificationResponse.ClassificationResults.Include;
}
return ClassificationResponse.ClassificationResults.Unchanged;
}
}And then we need to modify the function we call to accrue the interest thus:
ClassificationResponse isAccrued = await clsAccruedToday.Classify< InterestAccruedToday>();
if (isAccrued.Result == ClassificationResponse.ClassificationResults.Include )
{
// The accrual for today has been performed for this account
return req.CreateResponse<ProjectionFunctionResponse>(System.Net.HttpStatusCode.Forbidden,
ProjectionFunctionResponse.CreateResponse(startTime,
true,
$"Interest accrual already done on account {accountnumber} today",
isAccrued.AsOfSequence ),
FunctionResponse.MEDIA_TYPE
);
}Also because we are using this new classification we need to introduce concurrency protection in a try - catch block:
try
{
await bankAccountEvents.AppendEvent(evAccrued, isAccrued.AsOfSequence);
}
catch (EventSourcingOnAzureFunctions.Common.EventSourcing.Exceptions.EventStreamWriteException exWrite)
{
return req.CreateResponse<ProjectionFunctionResponse>(System.Net.HttpStatusCode.Forbidden,
ProjectionFunctionResponse.CreateResponse(startTime,
true,
$"Failed to write interest accrual event {exWrite.Message}",
projectedBalance.CurrentSequenceNumber),
FunctionResponse.MEDIA_TYPE);
}To this point all of the functions have run against a single bank account (event stream) only. However there is also the need to perform batch processes across all accounts - for example in a real bank the accrual of interest is performed as an overnight batch for all the bank accounts.
The first step is a timer triggered function that gets the account numbers of all of the bank accounts in the system and then passes them on to a durable functions orchestration that will do the processing.
// 1 - Accrue interest for all
// Triggered by a timer "0 30 1 * * *" at 01:30 AM every day
[FunctionName(nameof(AccrueInterestFoAllAccountsTimer))]
public static async Task AccrueInterestFoAllAccountsTimer(
[TimerTrigger("0 30 1 * * *",
RunOnStartup=false,
UseMonitor =true)] TimerInfo accrueInterestTimer,
[DurableClient] IDurableOrchestrationClient accrueInterestOrchestration,
[Classification("Bank", "Account", "ALL", @"")] Classification clsAllAccounts
)
{
// Get all the account numbers
IEnumerable<string> allAccounts = await clsAllAccounts.GetAllInstanceKeys();
// Pass them to the orchestration
await accrueInterestOrchestration.StartNewAsync(nameof(AccrueInterestForAllAccounts), allAccounts);
}This durable functions function then performs a fan out to perform the interest accrual function on all the accounts in parrallel.
// Accrue Interest For All Accounts
[FunctionName(nameof(AccrueInterestForAllAccounts))]
public static async Task AccrueInterestForAllAccounts
([OrchestrationTrigger] IDurableOrchestrationContext context)
{
IEnumerable<string> allAccounts = context.GetInput<IEnumerable<string>>();
if (null != allAccounts )
{
var accrualTasks = new List<Task<Tuple<string, bool>>>();
foreach (string accountNumber in allAccounts)
{
Task<Tuple<string, bool>> accrualTask = context.CallActivityAsync<Tuple<string, bool>>
(nameof(AccrueInterestForSpecificAccount), accountNumber );
accrualTasks.Add(accrualTask);
}
// Perform all the accruals in parallel
await Task.WhenAll(accrualTasks);
List<string> failedAccruals = new List<string>();
foreach (var accrualTask in accrualTasks)
{
if (! accrualTask.Result.Item2 )
{
failedAccruals.Add(accrualTask.Result.Item1);
}
}
// Try a second pass - using failedAccruals.Count ?
if (failedAccruals.Count > 0 )
{
throw new Exception("Not all account accruals succeeded");
}
}
}The action performed by each parallel call is very similar to the HTTP triggered version above:
//AccrueInterestForSpecificAccount
[FunctionName(nameof(AccrueInterestForSpecificAccount))]
public static async Task<Tuple<string, bool>> AccrueInterestForSpecificAccount
([ActivityTrigger] IDurableActivityContext accrueInterestContext)
{
const decimal DEBIT_INTEREST_RATE = 0.01M;
const decimal CREDIT_INTEREST_RATE = 0.005M;
string accountNumber = accrueInterestContext.GetInput<string>();
if (!string.IsNullOrEmpty(accountNumber))
{
EventStream bankAccountEvents = new EventStream(new EventStreamAttribute("Bank", "Account", accountNumber));
if (await bankAccountEvents.Exists() )
{
// Has the accrual been done today for this account?
Classification clsAccruedToday = new Classification(new ClassificationAttribute("Bank", "Account", accountNumber, nameof(InterestAccruedToday)));
ClassificationResponse isAccrued = await clsAccruedToday.Classify<InterestAccruedToday>();
if (isAccrued.Result != ClassificationResponse.ClassificationResults.Include)
{
// Get the account balance
Projection prjBankAccountBalance = new Projection(new ProjectionAttribute("Bank", "Account", accountNumber, nameof(Balance)));
// Get the current account balance, as at midnight
Balance projectedBalance = await prjBankAccountBalance.Process<Balance>(DateTime.Today);
if (null != projectedBalance)
{
Account.Events.InterestAccrued evAccrued = new Account.Events.InterestAccrued()
{
Commentary = $"Daily scheduled interest accrual",
AccrualEffectiveDate = DateTime.Today // set the accrual to midnight today
};
// calculate the accrual amount
if (projectedBalance.CurrentBalance >= 0)
{
// Using the credit rate
evAccrued.AmountAccrued = CREDIT_INTEREST_RATE * projectedBalance.CurrentBalance;
evAccrued.InterestRateInEffect = CREDIT_INTEREST_RATE;
}
else
{
// Use the debit rate
evAccrued.AmountAccrued = DEBIT_INTEREST_RATE * projectedBalance.CurrentBalance;
evAccrued.InterestRateInEffect = DEBIT_INTEREST_RATE;
}
try
{
await bankAccountEvents.AppendEvent(evAccrued, isAccrued.AsOfSequence);
}
catch (EventSourcingOnAzureFunctions.Common.EventSourcing.Exceptions.EventStreamWriteException exWrite)
{
// We can't be sure this hasn't already run...
return new Tuple<string,bool>(accountNumber, false);
}
}
}
}
}
return new Tuple<string, bool>(accountNumber, true);
}