Dekiru.Conduit 1.3.1

Conduits

Conduits is a minimal library for creating pipelines with middlewares. The intent is to use it encapsulate buisness logic in a way that is easy to overview and test.

A pipeline consists of an implementation of the Pipeline<TInput, TOutput>, with zero or more implementations of RequestInterceptor<TInput> and/or RequestResponseInterceptor<TInput, TOutput>.

TInput and TOutput may be the same type, or different types. The TInput of a pipeline is the type that is passed to the Process method of the pipeline. The TOutput of a pipeline is the type that is returned by the Process method of the pipeline. TInput must be a reference type, while TOutput may be a value type.

Note: TInput is considered the interface of a pipeline, and must be unique. Creating multiple pipelines with the same TInput is not supported, regardless of the TOutput.

There are two special abstractions for interceptors:

• GlobalInterceptor - Registered via the AddGlobalInterceptor extension method on IServiceCollection. These interceptors are called for all pipelines, in the order they are registered. They are called before the interceptors registered on the pipeline.
• ValidationInterceptor - Registered the same as any other interceptor, and accepts a type argument to identify the type to validate: AddInterceptor<ValidationInterceptor<MyClass>>.
  • These interceptors behave the same as other interceptors
  • They simplify using the Dekiru.FluentValidation library for validation, which in turn is built on top of FluentValidation.

Creating a pipeline

The following example demonstrates how to create a pipeline with a single interceptor.

public class MyPipeline : Pipeline<MyClass, MyClassResult>
{
	public MyPipeline()
	{
		AddInterceptor<MyInterceptor>(); // Add an interceptor with a parameterless constructor
		AddInterceptor(new MyInterceptorSingleton()); // Add a singleton interceptor
		AddInterceptor(provider => new MyInterceptor(provider.GetRequiredService<MyService>())); // Add an interceptor with a parameterized constructor, using a service from the DI container
		AddInterceptor(() => new MyInterceptor(1)); // Add an interceptor with a parameterized constructor
	}

	public override Task<MyClassResult> Process(MyClass input, CancellationToken cancellationToken)
	{
		return Task.FromResult(input);
	}
}

Things to note:

• The AddInterceptor methods can used to add interceptors to the pipeline. This is normally done in the constructor of the pipeline. The MyInterceptor class is an implementation of Interceptor<TInput, TOutput>.
• There are seven overloads of the AddInterceptor method:
  • AddInterceptor<TInterceptor>() - Adds an interceptor of type TInterceptor to the pipeline.
  • AddInterceptor(IRequestInterceptor<TInput> interceptor) - Adds an instance of TInterceptor to the pipeline.
  • AddInterceptor(Func<IServiceProvider, IRequestInterceptor<TInput>> factory) - Adds an interceptor to the pipeline using a factory method that receives an IServiceProvider.
  • AddInterceptor(Func<IRequestInterceptor<TInput>> factory) - Adds an interceptor to the pipeline using a factory method that does not receive any parameters.
  • AddInterceptor(IRequestResponseInterceptor<TInput, TOutput> interceptor) - Adds an instance of TInterceptor to the pipeline.
  • AddInterceptor(Func<IServiceProvider, IRequestResponseInterceptor<TInput, TOutput>> factory) - Adds an interceptor to the pipeline using a factory method that receives an IServiceProvider.
  • AddInterceptor(Func<IRequestResponseInterceptor<TInput, TOutput>> factory) - Adds an interceptor to the pipeline using a factory method that does not receive any parameters.
• The Process method is the entry point of the pipeline. This is where the input is processed by the pipeline.
• The Process method receives the input, and a CancellationToken.
• Any external services required by the pipeline should be declared as constructor parameters.
• Both the pipeline and the interceptors are registered as scoped services in the IServiceProvider.

Creating an interceptor

The following example demonstrates how to create an interceptor.

public class MyInterceptor : RequestResponseInterceptor<MyClass, MyClassResult>
{
	public override Task BeforeProcessing(MyClass input, PipelineContext context, CancellationToken cancellationToken)
	{
		return Task.CompletedTask;
	}

	public override Task AfterProcessing(MyClass input, MyClassResult output, PipelineContext context, CancellationToken cancellationToken)
	{
		return Task.CompletedTask;
	}

	public override Task OnError(MyClass input, List<Exception> exception, PipelineContext context, CancellationToken cancellationToken)
	{
		return Task.CompletedTask;
	}
}

Creating a RequestInterceptor<TInput> is similar, but without the TOutput type parameter, and no AfterProcessing method.

Things to note:

• Regardless of success or failure, the pipeline returns by calling the intercepters in the reverse order they were added to the pipeline.
  • If an exception occurs in an interceptor, the remaining interceptors are not called, but the OnError method is called for each interceptor in the reverse order.
• The BeforeProcessing and AfterProcessing methods have default implementations that return a completed task. These methods may be overridden to perform operations before and after the pipeline processes the input.
• The methods may modify, but not replace, the input and output of the pipeline.
• If an exception is thrown during the processing of the pipeline, the OnError method is called. This method may be overridden to perform operations when an exception occurs.
  • When an exception occurs, the AfterProcessing method is not called.
  • Regardless of errors thrown, each interceptor that had been called before the error occurred will have its OnError method called in the reverse order they were called.
  • One of the parameters of the OnError method is a list of exceptions. This list will contain all exceptions thrown by the pipeline and the interceptors that were called before the error occurred. An interceptor may add an exception to the list, or clear it.
  • Clearing the error list should be done carefully, as it will prevent the pipeline from throwing an error and will cause the pipeline to return the default value of TOutput.
• If the interceptor requires services from the DI container, they can be declared as constructor parameters.
• PipelineContext is a wrapper around a Dictionary<string, object> that can be used to store data that should be available to all interceptors in the pipeline.

Registering the pipelines

PipelineExtensions provides the extension method AddPipelines for IServiceCollection that registers all pipelines and interceptors in the assembly, as well as any referenced assemblies.

Pipelines are registered as Pipeline<TInput, TOutput> rather than the concrete implementation. This allows the pipelines to be resolved from the DI container based on the TInput type. The interceptors, instead, are registered via their implementation type, as the order of interceptors is important.

AddPipelines will also register the Dispatcher class as a singleton service. The Dispatcher class is used to dispatch the input to the correct pipeline based on the TInput type.

Dispatching the request

Dispatching a payload is done by resolving the Dispatcher class from the DI container and calling the DispatchAsync method.

var dispatcher = serviceProvider.GetRequiredService<Dispatcher>();

var result = await dispatcher.DispatchAsync<MyClass, MyClassResult>(new MyClass());

The pipeline is resolved based on the TInput type of the payload, rather then relying on the concrete implementation of the pipeline.

Attempting to dispatch a payload with an unregistered TInput type will result in an exception.

Abstractions

The library exposes a few abstractions, classes that can be used to simplify the implementation of pipelines and interceptors.

• ConcurrencyLimitInterceptor - An interceptor that limits the number of concurrent requests that can be processed by the pipeline.
• GlobalInterceptor - A base class for interceptors that should be called for all pipelines.
• ValidationInterceptor - A wrapper around the Dekiru.FluentValidation library that simplifies adding validation to a pipeline. Register the interceptor with the type to validate as a type argument: AddInterceptor<ValidationInterceptor<MyClass>> and the interceptor will find all validators for the type and run them.

Why?

The main goal of this library is to provide a way to encapsulate business logic in a way that is easy to overview and test, without introducing a lot of unnecessary code. By using pipelines and interceptors, the business logic can be split into smaller, more manageable parts. This also makes it easier to test the individual parts of the business logic.

Also, the use of Dekiru.FluentValidation provides an abstraction to quickly add validation to a pipeline, without having to write a lot of boilerplate code.

Why not use a library like MediatR? Honestly, mainly because MediatR contains a lot of features that we have no use for, as well as adding a bit more boilerplate code than we would like.

Its implementation of middleware handling relies on a bit of magic, which makes it harder to follow the flow of the code. By using a more explicit approach, we hope to make it easier to understand the code.