Dekiru.Conduit 8.0.3

Conduit

Conduit is a minimal library for creating pipelines with middlewares, and fire-and-forget events. 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> (or Pipeline<TInput>), with references to zero or more implementations of Interceptor<TInput, TOutput>.

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.

Exported types

• Dispatcher / IDispatcher - Used to dispatch a payload to the correct pipeline.
• EventHandler<T> - Base class for event handlers.
• Interceptor<TInput, TOutput> - Base class for interceptors.
  • GenericInterceptor - These interceptors are valid for all pipelines, but must be registered manually for each 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.
• IInterceptor<in TInput, in TOutput> - Interface for interceptors
  • The interface is contravariant, meaning that polymorphic dispatching is supported.
• Pipeline<TInput, TOutput> - Base class for pipelines.
• Pipeline<TInput> - Base class for pipelines without an output. Alias for Pipeline<TInput, Null>.
  • This kind of pipeline will still produce a result, but it will be Null, and any registered GenericInterceptor with a AfterProcessing method will be called.
• IResponse<T> - Interface that each request object must implement, identifying the type of the response.
  • IResponse - Alias for IResponse<Null>.
• Null - Helper class to represent a null value. Used as a return value for pipelines without an output.
  • Value - A static field that contains the only instance of Null.
  • Task - A static field that contains a completed task with the value of Null.
  • FromTask<T> - A static method that accepts (and resolves) a Task<T> and returns a Task<Null>.
  • FromTask - A static method that accepts a Task and returns a Task<Null>.
• PipelineContext - A wrapper around a Dictionary<string, object> that can be used to store data that should be available to all interceptors in the pipeline.
  • If you wish to add default values to the context, you can do so by registering PipelineContext as a scoped service and adding the values in the constructor.
• DoNotAutoRegisterAttribute - Used to prevent a class from being registered automatically by AddPipelines.

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
	}

	// Configure can be overriden and used to add interceptors, but it is not required
	// This is mainly useful to enable the use of primary constructors in .NET 8+
	protected override void Configure()
	{
		AddInterceptor<MyOtherInterceptor>(); // Add an interceptor with a parameterless constructor
	}

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

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 4 overloads of the AddInterceptor method:
  • AddInterceptor<TInterceptor>() - Adds an interceptor of type TInterceptor to the pipeline.
  • AddInterceptor(IInterceptor<TInput> interceptor) - Adds an instance of TInterceptor to the pipeline.
  • AddInterceptor(Func<IServiceProvider, IInterceptor<TInput>> factory) - Adds an interceptor to the pipeline using a factory method that receives an IServiceProvider.
  • AddInterceptor(Func<IInterceptor<TInput>> 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, a PipelineContext, 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(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;
	}
}

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.

Dispatching the request

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

Defining the request object:

public class MyClass : IResponse<MyClassResult>
{
}

Dispatching the request:

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

var result = await dispatcher.DispatchAsync(new MyClass());

The pipeline is resolved based on the TInput type of the payload, rather then relying on the concrete implementation of the pipeline. The dispatcher supports polymorphic dispatching, meaning that given the classes Foo and Bar, where Bar is a subclass of Foo, a pipeline defined for Foo will also handle Bar.

Attempting to dispatch a payload with type that has no matching pipeline will result in an exception.

Event handling

Events are nothing special in and of themselves, any C# class can be used as an event. Events are broadcast using the dispatcher's broadcast method:

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

var result = await dispatcher.Broadcast(new MyEventClass());

Any given class can have zero or more event handlers. Unhandled events will simply dropped quietly.

The event handler is implemented as follows:

public class MyEventHandler : EventHandler<MyEventClass>
{
    public override Task Handle(MyEventClass input, CancellationToken cancellationToken)
    {
		// Do event logic here
        return Task.CompletedTask;
    }
}

Events are purely fire and forget. Any exceptions raised by an event handler will be ignored. As such, each event handler must do its own error handling, retry functionality, etc.

NOTE: Event handling is still under development and the API and implementation details may change at any time.

Registering the pipelines, interceptor, and event handlers

PipelineExtensions provides the extension method AddPipelines for IServiceCollection that registers all pipelines, interceptors, and event handlers 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.

Event handlers are registered as IEventHandler<T> in the DI container.

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.

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