SubOrchestrations In Azure Durable Functions

And again, our topic is dedicated to Azure because it is one of the most popular cloud systems and if you’re a C#/.NET developer, in most cases, you will face to face with it.

So, what do we need to know before diving into our topic named “Suborchestration in Azure durable Functions: Practical approach”? Well, you need to read the below articles :

In our previous article, I talked about Orchestrator. In this section, we will talk about SubOrchestrator.

SubOrchestrator is an architectural approach that combines multiple activities in Azure functions. In this form of approach, activities are connected based on different patterns (chaining, fan-in fan-out, etc.) and are aimed at solving a single complex problem. SubOrchestrators usually appear within Orchestrators, during the division of labor. It links several activities and thus acts as a submodule for the main orchestrator.

In our task, we have allocated a separate suborchestrator for file transfer.

..................................
if(!getItemsResponse.DocumentListResponse.IsFolderEmpty()) {
    context.SetCustomStatus("Got source folder's documents. Preparing to move...");
    var rsp = new MoveFilesRequest {
        Request = getItemsResponse.DocumentListResponse,
            MoveFolderData = request
    };
    var _output = await context.CallSubOrchestratorAsyncstring >> ("HV_MoveFiles", rsp);
    output.AddRange(_output);
}
else {
    string message = "Folder Empty. Nothing to move!";
    context.SetCustomStatus(message);
    output.Add(message);
}.....................................

This SubOrchestrator is started only if it is possible to start the moving process and the directory to be moved to is not empty.

SubOrchestrators are invoked through the CallSubOrchestratorAsync method of the context (here IDurableOrchestrationContext). The Generic part shows what type of result the orchestrator will handle. In the argument part, it is stated which function will be started and with which parameter it will work. In the example, the HV_MoveFiles function is called by accepting an argument of type MoveFilesRequest.

var rsp = new MoveFilesRequest {
    Request = getItemsResponse.DocumentListResponse,
        MoveFolderData = request
};
var _output = await context.CallSubOrchestratorAsyncstring >> ("HV_MoveFiles", rsp);

SubOrchestrators, unlike regular activities, are marked with the OrchestratorTrigger attribute and receive a parameter of type IDurableOrchestrationContext because it is a separate subcontext by itself. This parameter is the main orchestrator and allows us to run activities within the function.

[FunctionName("HV_MoveFiles")]
public async Task < List < string > > MoveFiles([OrchestrationTrigger] IDurableOrchestrationContext context)

Durable functions accept a limited number of arguments (here it accepts only 1 argument), as we mentioned in our previous article. If there is a given argument to the subOrchestrator, we can get that parameter through the GetInput() method of the durable context.

var request = context.GetInput();

Then we identify the sub-folders in the main folder by creating a period equal to the number of files in the argument (folder) we receive. If there is a folder inside a folder, then we need to get the identifier of that folder by name. Therefore, we send the request to the service with activity.

if (request.GetItemInIndex(index).DisplayType == DisplayTypes.Folder) {
    string relativePath = request.GetDestinationItemPath(index);
    var subFolderExistsResponse = await context.CallActivityAsync("HV_GetFolderIdByName", new FolderRequest {
        RelativeFolderPath = relativePath,
            Id = request.Destination.BaseFolderId,
            SourceSystem = request.MoveFolderData.SourceSystem
    });...............................................

If it is possible to get the identifier of the folder, we call another activity and get the files inside that folder.

if (subFolderExistsResponse.HttpResponse.StatusCode == StatusCodes.Status200OK) {
    var subResponseBatch = await context.CallActivityAsync("HV_GetItemsInFolder", new GetItemsInFolderRequest {
        SourceSystem = request.MoveFolderData.SourceSystem, SourceFolderId = request.GetItemInIndex(index).Id, FolderName = request.GetItemInIndex(index).Name
    });.............................................

Then we transfer these files one by one from source to destination.

for (int subIndex = 0; subIndex < subResponseBatch.GetItemsCount(); subIndex++) {
    var moveDocumentData = new MoveDocumentData {
        SourceSystem = request.MoveFolderData.SourceSystem,
            MovedDocumentName = subResponseBatch.GetItemInIndex(subIndex).Name,
            DocumentType = subResponseBatch.GetItemInIndex(subIndex).DisplayType,
            Document = new Models.HomeVault.MoveDocument.Document {
                Destination = new Destination {
                        BaseFolderId = request.Destination.BaseFolderId, BaseFolderPath = request.Destination.BaseFolderPath, RelativeFolderPath = relativePath
                    },
                    MoveOptions = request.MoveFolderData.Document.MoveOption,
                    SourceDocumentId = subResponseBatch.GetItemInIndex(subIndex).Id
            }
    };
    var moveDocumentOutput = await context.CallActivityAsync("HV_MoveDocument", moveDocumentData);

Otherwise, if the searched element is not a folder, but a file, then we simply perform the transfer operation.

[FunctionName("HV_MoveFiles")]
public async Task < List < string > > MoveFiles([OrchestrationTrigger] IDurableOrchestrationContext context) {
    var output = new List < string > {
        "Move Files successfully started..."
    };
    var request = context.GetInput();
    try {
        for (int index = 0; index < request.GetItemsCount(); index++) {
            if (request.GetItemInIndex(index).DisplayType == DisplayTypes.Folder) {
                string relativePath = request.GetDestinationItemPath(index);
                var subFolderExistsResponse = await context.CallActivityAsync < GetFolderByIdOutput > ("HV_GetFolderIdByName", new FolderRequest {
                    RelativeFolderPath = relativePath,
                        Id = request.Destination.BaseFolderId,
                        SourceSystem = request.MoveFolderData.SourceSystem
                });
                output.AddRange(subFolderExistsResponse.Messages);
                if (subFolderExistsResponse.HttpResponse.StatusCode == StatusCodes.Status200OK) {
                    var subResponseBatch = await context.CallActivityAsync < DocumentListOutput > ("HV_GetItemsInFolder", new GetItemsInFolderRequest {
                        SourceSystem = request.MoveFolderData.SourceSystem, SourceFolderId = request.GetItemInIndex(index).Id, FolderName = request.GetItemInIndex(index).Name
                    });
                    output.AddRange(subResponseBatch.Messages);
                    bool isFolderEmpty = true;
                    if (subResponseBatch == null) {
                        string message = "Folder is empty. Nothing to move!";
                        output.Add(message);
                    } else {
                        for (int subIndex = 0; subIndex < subResponseBatch.GetItemsCount(); subIndex++) {
                            var moveDocumentData = new MoveDocumentData {
                                SourceSystem = request.MoveFolderData.SourceSystem,
                                    MovedDocumentName = subResponseBatch.GetItemInIndex(subIndex).Name,
                                    DocumentType = subResponseBatch.GetItemInIndex(subIndex).DisplayType,
                                    Document = new Models.HomeVault.MoveDocument.Document {
                                        Destination = new Destination {
                                                BaseFolderId = request.Destination.BaseFolderId, BaseFolderPath = request.Destination.BaseFolderPath, RelativeFolderPath = relativePath
                                            },
                                            MoveOptions = request.MoveFolderData.Document.MoveOption,
                                            SourceDocumentId = subResponseBatch.GetItemInIndex(subIndex).Id
                                    }
                            };
                            var moveDocumentOutput = await context.CallActivityAsync("HV_MoveDocument", moveDocumentData);
                            output.AddRange(moveDocumentOutput.Messages);
                            if (moveDocumentOutput.IsMoved) {
                                string message = $ "`{subResponseBatch.GetItemInIndex(subIndex).Name}` moved to `{relativePath}`.";
                                _logger.LogDebug(message);
                            } else {
                                string message = $ "`Warning : Can't move {subResponseBatch.GetItemInIndex(subIndex).Name}` to `{relativePath}`.";
                                _logger.LogError(message);
                                isFolderEmpty = false;
                            }
                        }
                    }
                    if (isFolderEmpty) {
                        output.Add($ "`{request.GetItemInIndex(index).Name}` folder is Empty. Preparing to delete empty folder...");
                        var _output = await context.CallActivityAsync("HV_DeleteDocument", new DeleteDocumentData {
                            SourceSystem = request.MoveFolderData.SourceSystem, FolderName = request.GetItemInIndex(index).Name, Id = request.GetItemInIndex(index).Id
                        });
                        output.AddRange(_output.Messages);
                    }
                } else {
                    var response = JsonConvert.DeserializeAnonymousType(subFolderExistsResponse.HttpResponse.Value.ToString(), new {
                        Message = ""
                    });
                    string message = $ "Warning : {response.Message}. Destination should have the folder before moving files!";
                    _logger.LogError(message);
                }
            } else
            if (request.GetItemInIndex(index).DisplayType == DisplayTypes.Document) {
                try {
                    var moveDocumentOutput = await context.CallActivityAsync("HV_MoveDocument", new MoveDocumentData {
                        SourceSystem = request.MoveFolderData.SourceSystem,
                            Document = new Models.HomeVault.MoveDocument.Document {
                                Destination = request.Destination,
                                    MoveOptions = request.MoveFolderData.Document.MoveOption,
                                    SourceDocumentId = request.GetItemInIndex(index).Id
                            }, MovedDocumentName = request.GetItemInIndex(index).Name, DocumentType = request.GetItemInIndex(index).DisplayType
                    });
                    output.AddRange(moveDocumentOutput.Messages);
                    if (moveDocumentOutput.IsMoved) {
                        string message = $ "`{request.GetItemInIndex(index).Name}` moved to `{request.Destination.BaseFolderPath}/{request.Destination.RelativeFolderPath}`.";
                        _logger.LogInformation(message);
                    } else {
                        string message = $ "`Can't move {request.GetItemInIndex(index).Name}`  to `{request.Destination.BaseFolderPath}/{request.Destination.RelativeFolderPath}`";
                        _logger.LogError(message);
                    }
                } catch (Exception exp) {
                    output.Add(exp.Message);
                }
            }
        }
    } catch (Exception exp) {
        output.Add(exp.Message);
        _logger.LogError(exp.Message);
    }
    return output;
}

To get items in a folder,

[FunctionName("HV_GetItemsInFolder")]
public async Task GetItemsInFolder([ActivityTrigger] GetItemsInFolderRequest request) {
    var output = new DocumentListOutput();
    output.AddMessage($ "Trying to get documents from `{request.FolderName}` folder...");
    try {
        output.DocumentListResponse = await _moveFolderProvider.GetItemsInFolderAsync(request.SourceSystem, request.SourceFolderId);
        output.AddMessage($ "{output.GetItemsCount()} {request.GenerateSuccessOutputMessage()}");
    } catch (Exception exp) {
        output.AddMessage(request.GenerateExceptionMessage(exp));
    }
    return output;
}

To get the ID of the folder, we run such an activity,

[FunctionName("HV_GetFolderIdByName")]
public async Task GetFolderIdByName([ActivityTrigger] FolderRequest request) {
    var output = new GetFolderByIdOutput();
    try {
        output.AddMessage($ "Preparing to find folder : `{request.SubFolderName}` inside Destination...");
        output.HttpResponse = await _moveFolderProvider.GetFolderIdByNameAsync(request.Id, request.RelativeFolderPath, request.SourceSystem);
        if (output.HttpResponse.StatusCode == StatusCodes.Status200OK) {
            output.AddMessage(request.GenerateSuccessOutputMessage());
        } else if (output.HttpResponse.StatusCode == StatusCodes.Status404NotFound) {
            output.AddMessage(request.GenerateExceptionMessage(null));
        }
    } catch (Exception exp) {
        output.AddMessage(request.GenerateExceptionMessage(exp));
    }
    return output;
}