Tag:

fan out/fan in

DevelopmentServerless

Durable Functions Workflows

by Mario Mamalis
written by Mario Mamalis

In this post we will examine the application of the workflow patterns covered in the Durable Functions Fundamentals part of this series. I will demonstrate a sample application, showcase how to setup the demo solution, and explain important parts of the code. I strongly suggest that you read the first post of this series before continuing here.

Solution Overview

The solution I developed for this demonstration is comprised by two separate applications. One is an Azure Durable Functions App, developed to run in Isolated Worker Mode and the other is a hosted Blazor WebAssembly App, used to visualize the workflow patterns. I used a hosted Blazor WebAssembly App because I wanted to create a SignalR Hub for real time updates to the user interface as the Functions run.

Together these applications make up the Coffee Shop Demo. Imagine a fictitious coffee shop where clients place their coffee orders at the register. After a coffee order is placed there are automated coffee machines that process the order and prepare the coffees, using the following steps: Grind, Dose, Tamp and Brew. All these operations depend on the coffee specifications. The coffee properties are 

  • Coffee Type: Espresso, Cappuccino and Latte
  • Intensity: Single, Double, Triple and Quad
  • Sweetness: None, Low, Medium and Sweet
Depending on the specs of each coffee, the automated coffee machine will execute the necessary steps and report the progress to the web application utilizing SignalR web sockets. Through this process we will be able to see how using the different workflow patterns affect the behavior and performance of the coffee machines.
 
Isolated Worker Mode

I would like to provide some insights about the decision to develop and run the Functions using the .NET isolated worker. The isolated worker enables us to run Functions apps using the latest .NET version. The other alternative we have, the Azure Functions In-Process mode, supports only the same .NET version as the Functions runtime. This means that only LTS versions of .NET are supported. At the time this solution was created .NET 7 was the latest version available but the Functions runtime was supporting .NET 6. 

Some of the benefits of using the isolated worker process are:
  • Fewer conflicts between the assemblies of the Functions runtime and the application runtime.
  • Complete control over the startup of the app, the configurations and the middleware.
  • The ability to use dependency injection and middleware.
Workflow Pattern Demonstrations
In the following short videos I will demonstrate the workflow patterns by running the applications and visualizing the differences.
Function Chaining
First we will take a look at the Function Chaining pattern. Using this pattern we will simulate a coffee order with 5 coffees of different type, intensity and sweetness. The Function Chaining Orchestrator will call four Activity Functions corresponding to the coffee making steps (Grind, Dose, Tamp and Brew), in sequential fashion one after the other for each coffee. In order to simulate the order entry I make an HTTP call to a Function Starter function which in turn calls the Orchestrator.

Fan out/fan in

Now we will take a look at the Fan out/fan in pattern. Using this pattern we will simulate a coffee order with the same 5 coffees we used before. This time however, the Fan Out Fan In Orchestrator will call the Coffee Maker Orchestrator for each coffee in the order in parallel. To do this I use the Coffee Maker Orchestrator as a sub-orchestrator. Again we will utilize Postman to make an HTTP call to the appropriate Starter Function. The difference in processing speed is clear! 

Note: I did not use the Chaining Orchestrator as a sub-orchestrator because the Chaining Orchestrator is a standalone Orchestrator that accepts a coffee order. The Coffee Maker Orchestrator accepts a coffee object and executes the steps for one coffee, which is more appropriate as a sub-orchestrator.

Human Interaction

The final demonstration will be the Human Interaction pattern. With this pattern we can pause the Durable Function Orchestrator at any point we want to receive an external event. The coffee order will be placed, but the coffee machine will not execute the order until the payment is received. To demonstrate that, I initiate the order and then after I get the appropriate prompt, I make another HTTP call using a predefined event and a specific URL for this purpose. The Human Interaction Orchestrator will intercept that event and will continue processing the order by calling the Coffee Maker sub-orchestrator.

Steps to Create the Durable Functions App

I used Visual Studio 2022 to create the Durable Functions App. The following snapshots demonstrate the process of creating the solution. Just click on them to enlarge.

Durable Functions App Solution Structure

After creating the solution, you will get the default files the template creates for you. Below is a snapshot of the solution in its final form after all the code was completed along with high level descriptions of the different classes.

Common Folder

Here you can find common Activity Functions, Models, SignalR classes, a Coffee Price Calculator and Enums.

Activity Functions

These are the functions that simulate the steps to prepare a coffee. Activity Functions is where you normally code your application logic. (Please view the first post in the series to see detailed information about different Function types).

Models

These are the POCO objects that represent the payloads passed in the Functions throughout the project. We have two objects. CoffeOrder and Coffee. The relationship is one-to-many where one CoffeeOrder can have many Coffees.
 

SignalR Classes

Here you will find all classes necessary to enable communication through SignalR. These classes are organized using composition and inheritance, using the appropriate SignalR client and exposing methods to send messages to the SignalR service I have provisioned on Azure. (I will not get into details about SignalR in this post as it is not the main topic).
 

Coffee Price Calculator

The CoffeePriceCalculator is a static class with static methods used to calculate the price of coffees and coffee orders.
 

Enums

The Enums class contains all enumerations used in the code such as the coffee type, sweetness and different statuses.

 

Patterns Folder

In this folder you will find all the Starters and Orchestrators, organized in subfolders for each pattern. As you can see we have Chaining, FanOutFanIn, and HumanInteraction subfolders containing the appropriate classes. 

The HumanInteraction subfolder contains 3 additional classes: An Activity Function (CalculateOrderPrice) that utilizes the Static CoffeePriceCalculator class to calculate the prices, another Activity Function (SendPaymentRequest) that simulates a delay of 2 seconds for sending a request for payment and a Model (ProcessPaymentEventModel) that is expected to be be included in the request made by the Human interacting with the orchestration. That model contains a property of PaymentStatus which is an enumeration. If the PaymentStatus passed to the event is “Approved”, then the orchestrator will continue the work and prepare the coffees.

Finally the CoffeeMaker subfolder contains the sub-orchestrator used by the FanOutFanIn and HumanInteraction Orchestrator Functions.

Important Code Sections

Now let’s go over some important code to help you understand how things come together. (The line numbers shown below are not the same as the line numbers on the actual code found on GitHub, for the obvious copy-paste reason).
 

Activity Functions

The code below is the Brew Activity Function. We can see that it has a constructor that accepts a CoffeeProcessStatusPublisher which is an implementation of the ICoffeeShopHubClient. As you can see we use Dependency Injection just like we would in any normal .NET 7 application. This is trivial at this point because we use the Isolated Worker Mode, as I explained in the beginning. This injected object is found in many other Functions and it is used to publish messages to SignalR so that we can visualize them eventually in the Blazor Web App.

You can see that the Activity Function has a name attribute. This  is very important because that is how we invoke Activity Functions from Orchestrators. We can also see that the Run method is an aync method and it has an ActivityTrigger attribute to indicate the type of Function it is. It accepts one object of type Coffee as the first parameter and then a FunctionContext as the second parameter. We can only pass a single input parameter to an Activity Function (other than the ontext). In this case we encapsulate all the properties we need in the POCO class Coffee and we are good to go. 

The FunctionContext is not used in this scenario but it could be used to call other Activity Functions if needed. If we were using the In-Process mode we would be using the Context to extract the input passed into the function since in that version parameters cannot be passed directly into the Function. The Context would be of a different type as well (IDurableActivityContext).

As mentioned above all the Activity Functions used in this project are simple and simulate the steps of the coffee making process. In a real scenario you can have more complex code. It is however recommended that you keep the function code simple and use well known design patterns to push the complexity in other class libraries.

public class Brew
{
    private readonly ICoffeeShopHubClient _coffeeProcessStatusPublisher;

    public Brew(ICoffeeShopHubClient coffeeProcessStatusPublisher)
    {
        _coffeeProcessStatusPublisher = coffeeProcessStatusPublisher;
    }

    [Function(nameof(Brew))]
    public async Task<Coffee> Run([ActivityTrigger] Coffee coffee, FunctionContext executionContext)
    {
        ILogger logger = executionContext.GetLogger(nameof(Brew));

        logger.LogInformation($"Brewing {coffee.Type} coffee  {coffee.Id} for order: {coffee.OrderId} ...");

        await Task.Delay((int)coffee.Intensity * 1000);

        coffee.Status = CoffeeStatus.Brewed;
        coffee.Message = "Brewing Completed";

        logger.LogInformation($"Brewing completed for {coffee.Type} coffee  {coffee.Id} of order: {coffee.OrderId}.");

        await _coffeeProcessStatusPublisher.UpdateCoffeeProcessStatus(coffee);

        return coffee;
    }
}

Starter Functions (Durable Clients)

The following code is a representation of a Starter Function otherwise known as a Durable Client Function. You can see that again we have a function name, a trigger (which in this case is an HttpTrigger) and two parameters: The DurableTaskClient and the FunctionContext.

This is one of the Functions I call using Postman using HTTP. The main purpose of a Durable Client Function is to kick of an Orchestrator Function. We can see that happening on lines 15 and 16. Before that, we can see that we read the request body and extract (deserialize) the CoffeOrder, which if you can remember from the videos, was passed in as a JSON object.

public static class ChainingStarter
{
    [Function(nameof(StartChainingOrchestrator))]
    public static async Task<HttpResponseData> StartChainingOrchestrator(
        [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
        [DurableClient] DurableTaskClient client,
        FunctionContext executionContext)
    {
        ILogger logger = executionContext.GetLogger("StartChainingOrchestrator");

        string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
        var coffeeOrder = JsonConvert.DeserializeObject<CoffeeOrder>(requestBody);

        // Function input comes from the request content.
        string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
            nameof(ChainingOrchestrator), coffeeOrder);

        logger.LogInformation("Started Coffee Order Orchestration with ID = '{instanceId}'.", instanceId);

        // Returns an HTTP 202 response with an instance management payload.
        // See https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-http-api#start-orchestration
        return client.CreateCheckStatusResponse(req, instanceId);
    }
}

Orchestrator Function: ChainingOrchestrator

In the first pattern demonstrated we utilize the ChainingOrchestrator. This first piece of code shows the constructor dependency injection, the function name and the appropriate trigger used (OrchestrationTrigger).

On line 14 you can see that I use a CreateReplaySafeLogger. If you rember from the first post of this series, there are many code constraints we have to follow in Orchestrator Functions because the Orchestrator stops and re-starts frequently. The code cannot create any ambiguity. To ensure reliable execution of the orchestration state, Orchestrator Functions must contain deterministic code; meaning it must produce the same result every time it runs. 

On line 15 I use the context to get the CoffeeOrder input parameter, however I could have passed it in as a parameter in the signature as well.

The code within the Try block starts the actual orchestration logic. I wanted to point out lines 29 and 36 as samples of how we call Activity Functions. You can see that we use the name of the function and we pass the input as a parameter.

public class ChainingOrchestrator
{
    private readonly ICoffeeShopHubClient _coffeeProcessStatusPublisher;

    public ChainingOrchestrator(ICoffeeShopHubClient coffeeProcessStatusPublisher)
    {
        _coffeeProcessStatusPublisher = coffeeProcessStatusPublisher;
    }

    [Function(nameof(ChainingOrchestrator))]
    public static async Task<CoffeeOrder> RunOrchestrator(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        ILogger logger = context.CreateReplaySafeLogger(nameof(ChainingOrchestrator));
        var coffeeOrder = context.GetInput<CoffeeOrder>();

        try
        {
            if (coffeeOrder == null)
            {
                coffeeOrder = new CoffeeOrder
                {
                    Status = OrderStatus.Failed,
                    Message = "Coffee order not specified."
                };

                logger.LogInformation(coffeeOrder.Message);

                await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

                return coffeeOrder;
            }

            coffeeOrder.Status = OrderStatus.Started;
            coffeeOrder.Message = $"Started processing coffee order {coffeeOrder.Id}.";
            await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

In the following code below (still in the same file), you can see how I chain the function calls. I iterate the coffee collection of the coffee order and invoke each Activity Function (coffee making step).

foreach (var coffee in coffeeOrder.Coffees)
{
    coffee.Message = $"Started making {coffee.Type} coffee {coffee.Id} for order {coffee.OrderId}.";
    await context.CallActivityAsync("UpdateCoffeeMakerStatus", coffee);

    var processedCoffee = await context.CallActivityAsync<Coffee>(nameof(Grind), coffee);
    processedCoffee = await context.CallActivityAsync<Coffee>(nameof(Dose), processedCoffee);
    processedCoffee = await context.CallActivityAsync<Coffee>(nameof(Tamp), processedCoffee);
    processedCoffee = await context.CallActivityAsync<Coffee>(nameof(Brew), processedCoffee);

    if (processedCoffee.Status == CoffeeStatus.Brewed)
    {
        processedCoffee.Message = $"{processedCoffee.Type} coffee {processedCoffee.Id} " +
            $"for order {processedCoffee.OrderId} is ready.";

        await context.CallActivityAsync("UpdateCoffeeMakerStatus", processedCoffee);
    }
}

Orchestrator Function: FanOutFanInOrchestrator

The following code is the most important part of the Fan Out Fan In pattern. First I create a List of Tasks of type Coffee (line 1). Then I iterate the coffee order and assign a Task to the return type of the sub-orchestrator function (CoffeeMakerOrchestrator). Each task returns a Coffee object that contains the status. Then on line 13 I await all Tasks to complete and when that is done the orchestrator finishes execution (a few steps below, not shown here).
 
This parallel execution of the sub-orchestrator makes the processing simulation prepare all coffees together. As you can see in the visualization in the video, the execution is a lot faster. The code of the sub-orchestrator is very similar to the Chaining Orchestrator code I showed above. The only difference is that the CoffeeMakerOrchestrator (used as a sub-orchestrator here), accepts a coffee instead of a coffee order. In this way I can do the parallel execution calls in the parent Orchestrator where they actually belong.

 

var coffeeTasks = new List<Task<Coffee>>();

coffeeOrder.Status = OrderStatus.Started;
coffeeOrder.Message = $"Started processing coffee order {coffeeOrder.Id}.";
await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

foreach (var coffee in coffeeOrder.Coffees)
{
    var task = context.CallSubOrchestratorAsync<Coffee>(nameof(CoffeeMakerOrchestrator), coffee);
    coffeeTasks.Add(task);
}

var coffeeMessages = await Task.WhenAll(coffeeTasks);

coffeeOrder.Status = OrderStatus.Completed;
coffeeOrder.Message = $"Coffee order {coffeeOrder.Id} is ready.";
await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

Orchestrator Function: HumanInteractionOrchestrator

This code below shows several interesting points of this pattern and the capability of the Durable Functions to wait for external events.

In line 1 you can see the call to the Activity Function that calculates the price. Immediately after the that I send a payment request and update the Coffee Order Status appropriately. On line 9 we see the code that sets the Orchestrator to sleep, waiting for an external event with name “ProcessPayment”, to trigger the rest of the code to execute. Once the event is received I examine the code, and only if the Payment Status is Approved, I proceed with the coffee making process.

After this point the code is identical to the Fan Out Fan in Pattern.

coffeeOrder = await context.CallActivityAsync<CoffeeOrder>("CalculateOrderPrice", coffeeOrder);

await context.CallActivityAsync("SendPaymentRequest", coffeeOrder);

coffeeOrder.Status = OrderStatus.Placed;
coffeeOrder.Message = $"Coffee order {coffeeOrder.Id} was placed. Total Cost: {coffeeOrder.TotalCost:C}. Waiting for payment confirmation...";
await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

var response = await context.WaitForExternalEvent<ProcessPaymentEventModel>("ProcessPayment");

if (response.PaymentStatus == PaymentStatus.Approved)
{
    coffeeOrder.Status = OrderStatus.Placed;
    coffeeOrder.Message = $"Payment received for coffee order {coffeeOrder.Id}";
    await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

    var coffeeTasks = new List<Task<Coffee>>();

    coffeeOrder.Status = OrderStatus.Started;
    coffeeOrder.Message = $"Started processing coffee order {coffeeOrder.Id}.";
    await context.CallActivityAsync("UpdateCoffeeOrderStatus", coffeeOrder);

    foreach (var coffee in coffeeOrder.Coffees)
    {
        var task = context.CallSubOrchestratorAsync<Coffee>(nameof(CoffeeMakerOrchestrator), coffee);
        coffeeTasks.Add(task);
    }

If you can remember from the corresponding video, when I called the Human Interaction HTTP endpoint with a POST, I received the response shown below. Line 4 in the response has the URL for posting events to this particular instance of the Orchestrator. If you scroll to the right you can see the {eventName} route parameter. I typed the name of the event there (ProcessPayment) and then made a POST call to the endpoint. At that point, this particular Orchestrator instance received the event and continued the code execution. This is pretty powerful!

{
    "id": "c7ccbd737d194f7cbf66995b8bcc3e03",
    "purgeHistoryDeleteUri": "https://func-durable-functions-demo.azurewebsites.net/runtime/webhooks/durabletask/instances/c7ccbd737d194f7cbf66995b8bcc3e03?code=7WrfquVTHSs9yXTv9Rk0-NacLhqEHrV2mozAHL-jYGTcAzFuA7_Erg==",
    "sendEventPostUri": "https://func-durable-functions-demo.azurewebsites.net/runtime/webhooks/durabletask/instances/c7ccbd737d194f7cbf66995b8bcc3e03/raiseEvent/{eventName}?code=7WrfquVTHSs9yXTv9Rk0-NacLhqEHrV2mozAHL-jYGTcAzFuA7_Erg==",
    "statusQueryGetUri": "https://func-durable-functions-demo.azurewebsites.net/runtime/webhooks/durabletask/instances/c7ccbd737d194f7cbf66995b8bcc3e03?code=7WrfquVTHSs9yXTv9Rk0-NacLhqEHrV2mozAHL-jYGTcAzFuA7_Erg==",
    "terminatePostUri": "https://func-durable-functions-demo.azurewebsites.net/runtime/webhooks/durabletask/instances/c7ccbd737d194f7cbf66995b8bcc3e03/terminate?reason={{text}}}&code=7WrfquVTHSs9yXTv9Rk0-NacLhqEHrV2mozAHL-jYGTcAzFuA7_Erg=="
}

Azure Resources

Both the Web Application and the Durable Functions Application shown during the demo have been running on resources provisioned on Azure. below you can see the resources used for both applications.

Conclusion

In this post I presented three important workflow patterns we can implement using Durable Functions. I hope this triggered your interest to investigate Durable Functions further. With Durable Functions we can develop complex workflows without the need of a messaging bus. Of course such a decision has pros and cons, but having all the Orchestration code organized in one place and not having to rely on other technologies is a pretty powerful argument. All the code I developed for this post can be found here: CoffeeShopDemo and DurableFunctionsDemo.
0 comment
0 FacebookTwitterPinterestEmail