Pipeline pattern: An assembly line of steps

You need to do a complex operation made of smaller consecutives tasks. These tasks might change from client to client. This is how you can use the Pipeline pattern to achieve that. Let’s implement the Pipeline pattern in C#.

With the Pipeline pattern, a complex task is divided into separated steps. Each step is responsible for a piece of logic of that complex task. Like an assembly line, steps in a pipeline are executed one after the other, depending on the output of previous steps.

TL;DR Pipeline pattern is like the enrich pattern with factories. Pipeline = Command + Factory + Enricher

When to use the Pipeline pattern?

You can use the pipeline pattern if you need to do a complex operation made of smaller tasks or steps. If a single task of this complex operation fails, you want to mark the whole operation as failed. Also, the tasks in your operation vary per client or type of operation.

Some common scenarios to use the pipeline pattern are booking a room, generating an invoice or creating an order.

Let’s use the Pipeline pattern

A pipeline is like an assembly line in a factory. Each workstation in an assembly adds a part until the product is assembled. For example, in a car factory, there are separate stations to put the doors, the engine and the wheels.

With the pipeline pattern, you can create reusable steps to perfom each action in your “assembly line”. Then, you run these steps one after the other in a pipeline.

For example, in an e-commerce system to sell an item, you need to update the stock, charge a credit card, send a delivery order and notify the client.

Let’s implement our own pipeline

First, create a command/context class for the inputs of the pipeline.

public class BuyItemCommand : ICommand
{
    // Item code, quantity, credit card information, etc
}

Then, create one class per each workstation of your assembly line. These are the steps.

In our e-commerce example, steps will be UpdateStockStep, ChargeCreditCardStep, SendDeliveryOrderStep and NotifyClientStep.

public class UpdateStockStep : IStep<BuyItemCommand>
{
    public Task ExecuteAsync(BuyItemCommand command)
    {
        // Put your own logic here
        return Task.CompletedTask;
    }
}

Next, we need a builder to create our pipeline with its steps. Since the steps may vary depending on the type of operation or the client, you can load your steps from a database or configuration files.

For our e-commerce example, we don’t need to create a delivery order when we sell an eBook. In that case, we need to build two pipelines: BuyPhysicalItemPipeline for products that require shipping and BuyDigitalItemPipeline for products that don’t.

But, let’s keep it simple. Let’s create a BuyItemPipelineBuilder.

public class BuyItemPipelineBuilder : IPipelineBuilder
{
    private readonly IStep<BuyItemCommand>[] Steps;

    public BuyItemPipelineBuilder(IStep<BuyItemCommand>[] steps)
    {
        Steps = steps;
    }

    public IPipeline CreatePipeline(BuyItemCommand command)
    {
      // Create your pipeline here...
      var updateStockStep = new UpdateStockStep();
      var chargeCreditCardStep = new ChargeCreditCard();
      var steps = new[] { updateStockStep, chargeCreditCardStep };
      return new BuyItemPipeline(command, steps);
    }
}

Now, create the pipeline to run all its steps. It will have a loop to execute each step.

public class BuyItemPipeline : IPipeline
{
    private readonly BuyItemCommand Command;
    private readonly IStep<BuyItemCommand>[] Steps;

    public BuyItemPipeline(BuyItemCommand command, IStep<BuyItemCommand>[] steps)
    {
        Command = command;
        Steps = steps;
    }

    public async Task ExecuteAsync()
    {
        foreach (var step in Steps)
        {
            await step.ExecuteAsync(Command);
        }
    }
}

Also, you can use the Decorator pattern to perform orthogonal actions on the execution of the pipeline or every step. You can run the pipeline inside a database transaction, log every step or measure the execution time of the pipeline.

Now everything is in place, let’s run our pipeline.

var command = new BuyItemCommand();
var builder = new BuyItemPipelineBuilder(command);
var pipeline = builder.CreatePipeline();

await pipeline.ExecuteAsync();

Some steps of the pipeline can be delayed for later processing. The user doesn’t have to wait for some steps to finish his interaction with the system. You can schedule the execution of some steps in background jobs for later processing. For example, you can use Hangfire or roll your own queue mechanism (Kiukie…Ahem, ahem)

Conclusion

Voilà! This is the Pipeline pattern. You can find it out there or implement it on your own. Depending on the expected load of your pipeline, you could use Azure Functions or any other queue mechanism to run your steps.

I have used and implemented this pattern before. I used it in an invoicing platform to generate documents. Each document and client type had a different pipeline.

Also, I have used it in a reservation management system. I had separate pipelines to create, modify and cancel reservations.

PS: You can take a look at Pipelinie to see more examples. Pipelinie offers abstractions and default implementations to roll your own pipelines and builders.

All ideas and contributions are more than welcome!