Pipelines and async

So the basic definition becomes:

public interface IAsyncPipelineStep<INPUT, OUTPUT>
{
	Task<OUTPUT> Process(Task<INPUT> input);
}

						

Based on that, we need to tweak the definitions of the abstract pipeline based on these steps, to follow the same pattern:

public abstract class AsyncPipeline<TIn, TOut> : IAsyncPipelineStep<TIn, TOut>
{
	public Func<Task<TIn>, Task<TOut>> _pipelineSteps { get; protected set; }

	public Task<TOut> ProcessAsync(Task<TIn> Input)
	{
		return _pipelineSteps(Input);
	}
}

						

And similarly adjust the helper method which allows us to compose the steps together:

public static class AsyncPipelineStepExtensions
{
	public static Task<TOut> Step<TIn, TOut>(this Task<TIn> Input, IAsyncPipelineStep<TIn, TOut> Step)
	{
		return Step.ProcessAsync(Input);
	}
}

						

That gives us a basis to try defining some steps... How about an async fetch for some HTML from a website?

public class HttpFetchAsyncStep : IAsyncPipelineStep<Uri, string>
{
	private static readonly HttpClient _client = new HttpClient();
	
	public async Task<string> ProcessAsync(Task<Uri> Input)
	{
		var uri = await Input;
		return await _client.GetStringAsync(uri);
	}
}

						

The interesting thing here is that our input to each step is a Task<INPUT> - that's the promise that there will be an instance of the INPUT class at some point in the future. That means our step needs to wait for that future-point before it does anything with the input. Hence the await Input being the first line here...

* sound of ominous foreshadowing *

And writing to disk can be async too:

public class DiskWriteAsyncStep : IAsyncPipelineStep<string, string>
{
	public async Task<string> ProcessAsync(Task<string> Input)
	{
		var data = await Input;
		var fileName = System.IO.Path.GetTempFileName();
		
		await System.IO.File.WriteAllTextAsync(fileName, data);
		
		return fileName;
	}
}

						

This is slightly different, because we're awaiting something which doesn't return data – but it's broadly the same pattern.

And we can compose those together into a pipeline:

public class ExampleAsyncPipeline : AsyncPipeline<Uri, string>
{
	public ExampleAsyncPipeline()
	{
		_pipelineSteps = input => input
			.Step(new HttpFetchAsyncStep())
			.Step(new DiskWriteAsyncStep());
		
	}
}

						

That doesn't change from our pre-async definition of a new pipeline. But we do need a little tweak to how we use it. Because our steps always have to take a Task<T> as input, we need a little tweak to ensure we have one for the input to the pipeline:

async Task Main()
{
	var pipeline = new ExampleAsyncPipeline();
	var uri = new Uri("https://news.bbc.co.uk/");

	var input = Task.FromResult(uri);

	var tempFile = await pipeline.ProcessAsync(input);

	Console.WriteLine($"{uri} saved to {tempFile}");
}

						

Now this runs, and it gives us some valid output:

https://news.bbc.co.uk/ saved to C:\Users\jeremy\AppData\Local\Temp\tmpF77A.tmp

						

But thinking about it, the code should probably hide the need for the Task.FromResult() call for us. We can amend the abstract AsyncPipeline to do this easily:

public abstract class AsyncPipeline<TIn, TOut> : IAsyncPipelineStep<TIn, TOut>
{
	public Func<Task<TIn>, Task<TOut>> _pipelineSteps { get; protected set; }

	public Task<TOut> ProcessAsync(Task<TIn> Input)
	{
		return _pipelineSteps(Input);
	}

	public Task<TOut> ProcessAsync(TIn Input)
	{
		var inputTask = Task.FromResult(Input);
		return _pipelineSteps(inputTask);
	}
}

						

Abstracting that away means we won't forget to do it later, and it makes the code to run the pipeline marginally neater too:

async Task Main()
{
	var pipeline = new ExampleAsyncPipeline();
	var uri = new Uri("https://news.bbc.co.uk/");

	var tempFile = await pipeline.ProcessAsync(uri);

	Console.WriteLine($"{uri} saved to {tempFile}");
}

						

And now all our pipelines can go async if necessary...

As previously, all the code above is in a gist if you want to tinker with it. (I've pasted that from LinqPad, so it's not a complete C# program, but it does include all the relevant code)

(And I want to add a "thank you!" to Walanem Figueiredo for asking the question that prompted me to stop and think about this – it's turned out to be a surprisingly interesting topic)

But!!! There's a problem here, I think url copied!

• Feel like sharing?
• ⇒ On BlueSky
• ⇒ On LinkedIn
• ⇒ On Mastodon
• ⇒ On Email