DEV Community: Robert

Pipeline pattern - intermittent flow code examples

Robert — Fri, 07 Jan 2022 09:01:30 +0000

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. 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.

Pipeline pattern - intermittent flow

Robert — Fri, 07 Jan 2022 08:58:19 +0000

Having implementation of pipeline pattern with error handling in place, let's start working on more sophisticated pipeline solution.

Let’s define additional requirements for pipeline implementation:

it allows to define an intermittent flow,
it allows to define step to read a message from the queue,
it allows to define step to send a message to a different queue,
it is terminated after sending out a message,
it allows to continue from a certain step after receiving response.

On the picture below you can find desired flow diagram:

The goal of ‘In Message’ step is to read a message from the queue. It means, message has to be delivered into this specific pipe step. In other words, we must be sure that message from the queue is part of defined pipe model. To ensure this, let’s create an interface which will contain such property:

public interface IInQueuePipeModel
{
    string InMessage { get; set; }
}

As you can see on flow diagram, it is possible to define multiple ‘In Message’ steps for reading messages from different queues. How does pipeline know which step should be invoked for a particular queue?

To receive a message from the queue or send out a message to some queue, it is required to know queue name. The queue name is key here and this is exactly what is going to be used to distinguish between different ‘In Message’ steps. So there are two things needed:

queue name present in pipe model to have information from which queue message was received,
queue name defined for specific 'In Message' step, so pipeline can figure out where is starting point for pipe execution.

To force queue name definition in pipe model, let’s define QueueName property in IInQueuePipeModel interface:

public interface IInQueuePipeModel
{
    string QueueName { get; set; }
    string InMessage { get; set; }
}

Additionally, separate interface for ‘In Message’ step is needed. It will ensure queue name presence for appropriate step:

public interface IInQueuePipeStep<TPipeModel> : IPipeStep<TPipeModel> where 
TPipeModel : IInQueuePipeModel
{
    string InQueueName { get; }
}

Notice I added generic type constraint on TPipeModel, meaning that the type parameter TPipeModel implements the IInQueuePipeModel interface.

Last part is ‘Out Message’ step. This step is going to be used for preparing request message for some external queue. Additionally, after this step, pipe is going to be terminated until receiving response.

To achieve that I need specific interface for ‘Out Message’ step which will indicate breaking point in the pipeline:

public interface IOutQueuePipeStep<TPipeModel> : IPipeStep<TPipeModel> where TPipeModel : IInQueuePipeModel
{ 
}

Also, I introduced yet another interface, so when needed, pipe model will implement property for outgoing messages:

public interface IOutQueuePipeModel
{
    string OutMessage { get; set; }
}

IOutQueuePipeModel interface is not required. Since there are different ways to implement sending out messages, such property is not always needed in pipe model. Add it when you are sure you will make use of it.

Even though whole pipeline is defined as one piece, in runtime it will be split into multiple executions, as flow diagram shows. In other words there is going to be a mechanism in place that will extract only steps that are required for a particular execution. How to achieve that? Let's make use of interfaces we've already defined: IInQueuePipeStep and IOutQueuePipeStep. Additionally, new QueuePipeService will be introduced and it's going to be responsible for queue related pipe execution.

Below you can find implementation of QueuePipeService. The only difference comparing to the previous implementation is inside ExecuteAsync method:

public class QueuePipeService<TPipeModel> : IPipeService<TPipeModel>
{
    protected readonly IList<Func<IPipeStep<TPipeModel>>> _pipeSteps;

    protected QueuePipeService()
    {
        _pipeSteps = new List<Func<IPipeStep<TPipeModel>>>();
    }

    public IPipeService<TPipeModel> Add(Func<IPipeStep<TPipeModel>> pipeStep)
    {
        _pipeSteps.Add(pipeStep);
        return this;
    }

    public async Task ExecuteAsync(TPipeModel pipeModel)
    {
        var steps = GetInitializedSteps().SkipUntil(s => !(s is IInQueuePipeStep<TPipeModel> && (s as IInQueuePipeStep<TPipeModel>).InQueueName == pipeModel.QueueName))
                                         .TakeUntil(s => !(s is IOutQueuePipeStep<TPipeModel>));

        foreach (var pipeStep in steps)
        {
            await pipeStep.ExecuteAsync(pipeModel);
        }
    }

    private IEnumerable<IPipeStep<TPipeModel>> GetInitializedSteps()
    {
        return _pipeSteps.Select(s => s.Invoke());
    }
}

ExecuteAsync works in a way that until it finds step which implements IInQueuePipeStep interface and queue name for such step is equal to queue name passed in pipe model it will be skipping steps. From the matching step it gets all next steps but if one of them implements IOutQueuePipeStep, it is going to be an indication that after this step some request message is going to be sent out and pipeline will be terminated until receiving response. To achieve this functionality I created two IEnumerable extension methods: SkipUntil and TakeUntil.

As I mentioned earlier, there is a big chunk of code which is common for both: QueuePipeService and PipeService. To follow DRY (don’t repeat yourself) principle, under this link, you can find final implementation with extracted PipeServiceBase abstract class which contains common implementation for both services.

Based on the final implementation, PipelinePattern NuGet package is generated. It is ready to be downloaded and used.

Summary

Using proposed pipe implementation, whole business process can be defined in a single place. Moreover, implementation of complex business flow will have clean, SOLID, testable and easy to understand code. Without knowing implementation details, looking just on pipe steps, at the first glance it is quite obvious what is going to happen in specific pipeline. It is just a matter of digging into specific step to find out everything. Additionally, if you integrate logging into your pipeline, issues can be tracked easily.

I hope you will make use of it in your project.

You can find it in NuGet package manager by typing ‘PipelinePattern’.

Future enhancement

There are more things that could be incorporated into pipeline solution, like for instance:

parallel execution of a series of steps,
reporting of pipeline progress with SignalR.

Stay tuned!

Pipeline pattern - implementation

Robert — Mon, 20 Dec 2021 12:14:15 +0000

Introduction

Pipeline pattern implementation – so popular topic. A lot of different articles were already created to cover different aspects of this pattern. But have you ever tried to incorporate it into very complicated flow with usage of Azure Functions or WebJob and still having both: clean, SOLID, testable code and pipeline implementation which can be interrupted to perform some external request and be able to get back to appropriate step when notified and proceed further?

Let me share with you my point of view on this topic. In this series of articles you will find all implementation details with explanation and examples. Let’s get started!

Implementation

First, the pipeline pattern itself will be implemented and after that I will develop it further.

Before writing the code, I would like to define all the assumptions the final solution should meet:

Pipeline pattern implementation is generic.
Implementation follows SOLID principles.
Error handling is in place.

Let’s begin with first point.

If you are new to this pattern, before continue reading, stop here for a while and ask yourself couple of questions:

How should my pipeline pattern implementation work?
What is the input/output for my pipeline?
How would I define generic implementation of pipeline pattern?
What would I need to make it generic?
What methods should be provided by generic implementation?
How am I going to implement pipe step?

Are you ready with your ideas? Below are mine.

Pipeline requires steps. Step is nothing but separate class with single responsibility. Each step requires model to interact with when its logic is executed. So here is starting point:

public interface IPipeStep<TPipeModel>
{
    Task ExecuteAsync(TPipeModel pipeModel);
}

On top of that, pipe service for managing pipe steps is needed.

Its interfaces: IPipeService with method definition to set up pipeline and IPipeServiceExecution to define method for pipe execution:

public interface IPipeService<TPipeModel> : IPipeServiceExecution<TPipeModel>
{
    IPipeService<TPipeModel> Add(Func<IPipeStep<TPipeModel>> pipeStep);
}

public interface IPipeServiceExecution<TPipeModel>
{
    Task ExecuteAsync(TPipeModel pipeModel);
}

And its implementation:

public class PipeService<TPipeModel> : IPipeService<TPipeModel>
{
    private readonly IList<Func<IPipeStep<TPipeModel>>> _pipeSteps;

    public PipeService()
    {
        _pipeSteps = new List<Func<IPipeStep<TPipeModel>>>();          
    }

    public IPipeService<TPipeModel> Add(Func<IPipeStep<TPipeModel>> pipeStep)
    {
        _pipeSteps.Add(pipeStep);
        return this;
    }

    public async Task ExecuteAsync(TPipeModel pipeModel)
    {
        foreach (var pipeStep in _pipeSteps)
        {
            await pipeStep.Invoke().ExecuteAsync(pipeModel);
        }
    }
}

That’s it! Simple, isn’t it? Generic pipe implementation is ready to be used. Let’s create first pipeline.

Wait...create pipeline? What is the best place to do it? Most probably it's going to have multiple steps, right? Each step is going to be represented by separate class. Steps must be combined together and at the very end IPipeServiceExecution should be returned. Isn’t it a great place for factory? Indeed, it is!

Let’s create yet another generic interface for pipeline factory which will be utilized for concreate pipe factory implementation. Factory is going to have single responsibility – create pipeline ready for being executed. Factory always provides complete pipeline with all defined steps. Also what is really important, CreatePipe method returns IPipeServiceExecution interface, so the only action that can be performed when pipe is returned from factory is its execution.

public interface IPipeFactory<TPipeModel>
{
    IPipeServiceExecution<TPipeModel> CreatePipe();
}

You can find complete, generic pipeline implementation here.

Using this link, you can navigate to an example of “Text transformation” pipeline which uses above implementation. As an input it takes unformatted text and based on defined steps it transforms it into desired format.

Notice how easy we can test each step. Additionally, each step has always single responsibility. We also have open-closed principle in place. It all means that 2nd point from our assumptions list is already fulfilled – we do follow SOLID principles.

Just additional side note: remember, steps order really matters in pipeline!

What about error handling?

Of course you can do it in each and every step or catch it outside of pipeline execution. But what if it would be required to define multiple error handling steps with the need of using pipeline model for handling an issue correctly?

Again, keeping in mind single responsibility principle and unit testing let's extend IPipeService interface to have a possibility to add error handling steps which will be executed in case pipe breaks:

public interface IPipeService<TPipeModel> : IPipeServiceExecution<TPipeModel>
{
    IPipeService<TPipeModel> Add(Func<IPipeStep<TPipeModel>> pipeStep);
    IPipeService<TPipeModel> AddErrorStep(Func<IPipeStep<TPipeModel>> errorStep);
}

Final implementation of pipeline pattern with error handling can be found here. Code example is available under this link.