Dekiru.Conduit 1.1.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 Interceptor<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 endpoint of a pipeline, and must be unique. Creating multiple pipelines with the same TInput is not supported, regardless of the TOutput.

There is a special abstract variant of the interceptor, called GlobalInterceptor. This interceptor is called for every pipeline, regardless of the TInput type. This can be useful for logging, error handling, or other global concerns. Use the extension method AddGlobalInterceptor method to register a global interceptor. They are called in the order they are added, and before any pipeline-specific interceptors, but are otherwise treated the same as regular interceptors.

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 four overloads of the AddInterceptor method:
  • AddInterceptor<TInterceptor>() - Adds an interceptor of type TInterceptor to the pipeline.
  • AddInterceptor(IInterceptor<TInput, TOutput> interceptor) - Adds an instance of TInterceptor to the pipeline.
  • AddInterceptor(Func<IServiceProvider, IInterceptor<TInput, TOutput>> factory) - Adds an interceptor to the pipeline using a factory method that receives an IServiceProvider.
  • AddInterceptor(Func<IInterceptor<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 : Interceptor<MyClass, MyClassResult>
{
	public override Task BeforeProcessing(MyClass input, PipelineContext context, CancellationToken cancellationToken)
	{
		return Task.CompletedTask;
	}

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

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

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

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.

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.

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.