Pipeline pattern - intermittent flow code examples
Tags: csharp, azure, codequality
Published: 2022-01-07 09:01:30

## Pipeline example using Azure Functions
Let’s imagine a request to translate header, body and footer of some article. But before marking translation as accepted, it has to be approved by some external unit as well. To get approval, request message for text acceptance must be placed on some external queue. When approval response is received, pipeline can be finalized.

Working example can be found [here](https://github.com/rwasik/pipeline-pattern-examples/tree/pipeline-azure-functions). To run it, all you have to do is to define your ServiceBus connection string and create three queues:
- translate-text
- approve-text-out
- approve-text-in

Using for instance ServiceBusExplorer tool you can put a messages on the queues.

Couple words about implementation.

`TranslateTextPipeModel` implements two interfaces:
- `IInQueuePipeModel`
- `IOutQueuePipeModel`

There are two ‘In Message’ steps:
- `DeserializeTranslateTextRequestStep`
- `DeserializeApprovalTextResponseStep`

and one ‘Out Message’ step:
- `PrepareApprovalTextRequestStep`

I have defined `TranslateTextPipeFactory` which is used inside `TranslateTextService` to create complete pipeline. Moreover, in service, pipe model with queue name and input message is initiated:
```
var pipeModel = new TranslateTextPipeModel
{
    QueueName = queueName,
    InMessage = message
};
```
Service also invokes pipeline execution.

Two Azure Functions are defined:
- `TranslateTextFunction`
- `TextApprovalFunction`

Both functions make use of `TranslateTextService`. First function receives a message from “translate-text” service bus and at the end it puts a request for text approval on “approve-text-out” queue. With Azure Functions that return value, you can bind a function output binding to the return value. So if you want return value to be put on service bus, you have to specify following attribute:
```
[return: ServiceBus("approve-text-out", Connection = "ServiceBus")]
```
Second function is invoked when text approval response is delivered to “approve-text-in” queue. This function finalizes the pipeline flow.

Simple, isn’t it?

## Pipeline example using Azure WebJob

Example used for WebJob is exactly the same example as for Azure Functions – request for header, body and footer translation of some article.

Solution for Azure Web Job is a bit different. First of all a  message must be put on external queue by ourselves. Using Azure Functions it was possible to apply the output binding attribute.

When it comes to receiving messages, it can be done by simply defining functions desired for it or by making use of `IHostedService` interface to implement some background queue related tasks.

I will present you second approach. Using `IHostedService` interface, `ServiceBusListener` for pipeline will be implemented. For each pipeline queue that delivers messages, message handler is going to be registered automatically. Thanks to that it's possible to define general flow for receiving messages and passing them into pipeline execution.

If by any mean it is required to extend pipeline and add some additional external call, all what is needed is just to define additional queue name and pipeline steps to handle this specific case.

As I've already mentioned, sending message to external queue must be handled by ourselves. For that let’s define generic `OutServiceBusPipeStep` which will implement `IOutQueuePipeStep` interface. Generic step implementation will contain three abstractions required to be overridden by the inheriting class:
- `ServiceBusConnectionConfigKey` (property)
- `OutQueueName` (property)
- `CreateOutQueueMessage(TPipeModel pipeModel)` (function)

`ExecuteAsync` function from `OutServiceBusPipeStep` will prepare message, create `QueueClient` and based on the configuration passed through properties it will put a message on the queue.

From now on this generic implementation can be used for all WebJob pipe steps that are responsible for sending message to external queue.

Next part that differs from Azure Functions solution is the way of handling incoming messages.

For that I defined abstract `ServiceBusListenerBase<TPipeModel, TInQueueEnum>` class which implements `IHostedService` interface. Implementing such interface it is required to implement following  interface members:
- `Task StartAsync(CancellationToken cancellationToken)`
- `Task StopAsync(CancellationToken cancellationToken)`

`StartAsync` method will iterate through all queues that deliver messages for pipeline and for each and every queue, message handler will be registered. Message handler does nothing but receiving message, preparing pipe model, creating and executing pipeline.

Has the second generic type parameter for `ServiceBusListenerBase` caught your attention?

To have single source of truth for all queues related to receiving or sending messages for specific pipeline I created two enumeration types:
- `TranslateTextPipeInQueue`
- `TranslateTextPipeOutQueue`

Enumeration type which defines ‘InQueue’ values - `TranslateTextPipeInQueue`, will be passed as `TInQueueEnum` type into `ServiceBusListenerBase`. Thanks to that, I’m able to iterate through its values and register message handler for defined queues.

Another benefit of introducing an enumeration type is that this is the only place to be modified if new queue will become part of existing process.

Final solution can be found under [this link](https://github.com/rwasik/pipeline-pattern-examples/tree/pipeline-azure-webjob).