Handling boilerplate tasks via service pipelines

Question

Here's a small experiment about handling boilerplate tasks via service pipelines. It's similar to the HTML pipeline but simpler and multipurpose.

---

I've updated the question and this is the 2nd version of this project. The first one turned out to have some substantial weaknesses like inability to invoke more than one controller service to handle requests. The code is written in C# 10 and .net-6.

---

Example

Let's start with a demo-app demonstrating how I'd use it:

public class ServicePipelineDemo2
{
    public static async Task Test()
    {
        // Compose the container.
        var builder = new ContainerBuilder();
        
        // Register cache dependency,
        builder.RegisterInstance(new MemoryCache(new MemoryCacheOptions())).As<IMemoryCache>().SingleInstance();

        // Register service-pipeline-builder and associate it with ReadFile.Text by name.
        builder.Register(c => new Service.PipelineBuilder
        {
            // This pipeline should resolve environment variables that might be used by the .Name property.
            new EnvironmentVariableService(PropertyService.For<IReadFile>.Select(x => x.Name)),
            // All text files should be cached for 30min and other files for 15min.
            new CacheLifetimeService(TimeSpan.FromMinutes(15))
            {
                Rules =
                {
                    // Add a condition for read-file request.
                    Condition.For<IReadFile>.When(x => x.Name.EndsWith(".txt"), ".txt").Then(TimeSpan.FromMinutes(30))
                }
            },
            new CacheService(c.Resolve<IMemoryCache>(), PropertyService.For<IReadFile>.Select(x => x.Name)),
            // This pipeline should look in embedded resources first before it reads files. 
            // Since this node doesn't have to succeed, the next one is called.
            new EmbeddedResourceService<ServicePipelineDemo2> { MustSucceed = false },
            // This overrides the file-service for testing.
            new ConstantService.Text("This is not a real file!"),
            // Finally this node tries to read a file.
            new FileService.Read()
        }).InstancePerDependency().Named<Service.PipelineBuilder>(nameof(ReadFile.Text));

        await using var container = builder.Build();
        await using var scope = container.BeginLifetimeScope();

        // Set some environment variable.
        Environment.SetEnvironmentVariable("HOME", @"c:\temp");

        // Create the request and invoke it.
        var result = await new ReadFile.Text(@"%HOME%\notes.txt").CacheLifetime(TimeSpan.FromMinutes(10)).InvokeAsync(scope);

        Console.WriteLine(result); // --> "This is not a real file!" or FileNotFoundException (as there is no "notes.txt".
    }
}

FAQ

• Why do I use new ReadFile.Text instead of resolving the serivce via dependency injection? -- The constructor allows me to better enforce required parameters. Since this is only a request and doesn't require any dependencies here, I'd say it's fine. Additionally it also specifies the expected return type.
• How do I associate pipelines with requests? -- I use Named services. By default the name of the request type is used (like Text here). Such a pipeline would apply to all requests of this type but it's possible to add a Tag to the Items dictionary and use this for the lookup.
• Why do I register the pipeline as a Func<>? -- It should be possible to create independent pipelines for each request but at the end it depends on the actual use case so there can be a single pipeline for a lifetime scope.
• In which order are the middleware evaluated? -- They are evaluated in the same order as registered.
• Why Autofac? -- I like Autofac. Microsoft's DI is too limited.
• Is there a GitHub link? -- There is.
• What would be some other usage scenarios?
  • To completely replace the middleware with something else for testing and throw exceptions or return other results without modyfing the requesting code.
  • To add another middleware to create file copies or backups without modyfing the code where WriteFile is requested.
  • To add validation middleware.
  • To add json de/serialization middleware.
  • To create an email-service and middleware attaching files or signatures.
  • To create a config-service that today might read settings from appsettings.json and tommorow from a database by only chanigng the middleware. Another middleware could at the same time validate settings.
  • To add a telemetry middleware that would log the usage of each request.
  • To add multiple services of the same type and try to read a file from different locations and use the first successful result.
  • To add a retry middleware.
  • ...

Core

Like the HTTP pipeline there is also a request but no context and response objects. The result is returned directly. The request specifies what you'd like to do e.g.: ReadFile or CopyFile. It travels through the pipeline where each node can do anything with it.

Nodes or middleware are called ISerivce. They point to the next one and have one additional property MustSucceed. By default this is false for normal nodes and true for controller nodes like ReadFile when it's the last one in the chain and a result must be returned or an exception should be thrown.

The Items dictionary allows to pass additional data.

public interface IRequest
{
    IDictionary<string, object> Items { get; }
}

public interface IRequest<T> : IRequest { }

public abstract class Request<T> : IRequest<T>
{
    public IDictionary<string, object> Items { get; } = new Dictionary<string, object>(SoftString.Comparer);
}

public sealed class Unit
{
    private Unit() { }

    public static readonly Unit Default = new();
}

public interface IService : IEnumerable<IService>
{
    bool MustSucceed { get; }

    // Points to the next middleware.
    IService? Next { get; set; }

    Task<object> InvokeAsync(IRequest request);
}

public abstract class Service : IService
{
    public bool MustSucceed { get; set; }

    public IService? Next { get; set; }

    public abstract Task<object> InvokeAsync(IRequest request);

    public IEnumerator<IService> GetEnumerator() => this.Enumerate().GetEnumerator();

    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();

    protected async Task<object> InvokeNext(IRequest request)
    {
        return
            Next is { } next
                ? await next.InvokeAsync(request)
                : Unit.Default;
    }

    private class Empty : Service
    {
        public override async Task<object> InvokeAsync(IRequest request) => await InvokeNext(request);
    }

    public class PipelineBuilder : IEnumerable<IService>
    {
        private IService First { get; } = new Empty();

        // Adds the specified service at the end of the pipeline.
        public void Add(IService last) => First.Enumerate().Last().Next = last;

        public IEnumerator<IService> GetEnumerator() => First.Enumerate().GetEnumerator();

        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();

        public IService Build() => First;
    }
}

The Service also provides the PipelineBuilder that knows how to connect serivces that are added to it.

Implementation

For the sake of this experiment I created two ReadFile requests:

• Text gets a string
• Stream gets a FileStream.

public interface IReadFile : IRequest
{
    public string Name { get; set; }

    public FileShare Share { get; }
}

public abstract class ReadFile<T> : Request<T>, IReadFile
{
    protected ReadFile(string name) => Name = name;

    public string Name { get; set; }

    public FileShare Share { get; set; } = FileShare.Read;
}

public abstract class ReadFile
{
    public class Text : ReadFile<string>
    {
        public Text(string name) : base(name) { }

        public Encoding Encoding { get; set; } = Encoding.UTF8;
    }

    public class Stream : ReadFile<FileStream>
    {
        public Stream(string name) : base(name) { }
    }
}

Middleware

As far as reading files is concered it's common to perform such tasks as:

• resolving environment variables in paths or
• caching results.

Both of them are handled by two middleware serivces. They work dynamically with properties specified via the constructor. (The cache-service is a dummy but the environment-variable service updates the file-name.)

public class EnvironmentVariableService : Service
{
    public EnvironmentVariableService(IPropertyAccessor<string> property) => Property = property;

    private IPropertyAccessor<string> Property { get; }

    public override async Task<object> InvokeAsync(IRequest request)
    {
        Property.SetValue(request, Environment.ExpandEnvironmentVariables(Property.GetValue(request)));

        return await InvokeNext(request);
    }
}

public class CacheService : Service
{
    public CacheService(IMemoryCache cache, IPropertyAccessor<string> key) => (Cache, Key) = (cache, key);

    private IMemoryCache Cache { get; }

    private IPropertyAccessor<string> Key { get; }

    public override async Task<object> InvokeAsync(IRequest request)
    {
        if (request.CacheLifetime() is var cacheLifetime && cacheLifetime > TimeSpan.Zero)
        {
            Console.WriteLine($"Cache-lifetime: {cacheLifetime}");
        }

        return await InvokeNext(request);
    }
}

// Allows to associate cache-lifetime to with a service.
public class CacheLifetimeService : Service
{
    public CacheLifetimeService(TimeSpan fallback)
    {
        if (fallback == TimeSpan.Zero) throw new ArgumentException("Fallback value needs to be greater than zero.");
        Fallback = fallback;
    }

    private TimeSpan Fallback { get; }

    public List<ConditionBag<TimeSpan>> Rules { get; } = new();

    public override async Task<object> InvokeAsync(IRequest request)
    {
        var lifetime = Rules.Where(c => c.Evaluate(request)).Select(c => c.GetValue()).FirstOrDefault();

        request.CacheLifetime(lifetime > TimeSpan.Zero ? lifetime : Fallback);

        return await InvokeNext(request);
    }
}

Controllers

Serives that handle the ReadFile request are:

• FileService and
• EmbeddedResourceService

I call them controllers although they are like every other middleware and can be placed anywhere in the pipeline.

public abstract class FileService : Service
{
    protected FileService()
    {
        MustSucceed = true;
    }

    public class Read : FileService
    {
        public override async Task<object> InvokeAsync(IRequest request)
        {
            if (request is IReadFile file)
            {
                if (File.Exists(file.Name))
                {
                    if (request is ReadFile.Text t)
                    {
                        await using var stream = new FileStream(file.Name, FileMode.Open, FileAccess.Read, file.Share);
                        return await stream.ReadTextAsync(t.Encoding);
                    }

                    if (request is ReadFile.Stream s)
                    {
                        return new FileStream(file.Name, FileMode.Open, FileAccess.Read, file.Share).ToTask();
                    }
                }
                else
                {
                    return
                        MustSucceed
                            ? throw DynamicException.Create("FileNotFound", $"There is no such file as '{file.Name}'.")
                            : await InvokeNext(request);
                }
            }

            throw DynamicException.Create("UnknownRequest", $"{request.GetType().ToPrettyString()} is not supported by this {nameof(FileService)}.");
        }
    }
}

public class EmbeddedResourceService : Service
{
    public EmbeddedResourceService(Assembly assembly)
    {
        Assembly = assembly;
        MustSucceed = true;
    }

    private Assembly Assembly { get; }

    public override async Task<object> InvokeAsync(IRequest request)
    {
        if (request is IReadFile file)
        {
            var name = Normalize(file.Name);
            if (FindResource(name) is { } stream)
            {
                if (request is ReadFile.Text t)
                {
                    await using (stream)
                    {
                        return await stream.ReadTextAsync();
                    }
                }

                if (request is ReadFile.Stream)
                {
                    return stream;
                }
            }
            else
            {
                Console.WriteLine($"Embedded resource '{name}' not found.");
                
                return
                    MustSucceed
                        ? throw DynamicException.Create("EmbeddedResourceNotFound", $"There is no such file as '{file.Name}'.")
                        : await InvokeNext(request);
            }
        }

        throw DynamicException.Create("UnknownRequest", $"{request.GetType().ToPrettyString()} is not supported by this {nameof(EmbeddedResourceService)}.");
    }

    // Embedded resource names are separated by '.' so replace the windows separator.
    private static string Normalize(string name) => Regex.Replace(name, @"\\|\/", ".");

    private Stream? FindResource(string name)
    {
        // Embedded resource names are case sensitive so find the actual name of the resource.
        var actualName = Assembly.GetManifestResourceNames().FirstOrDefault(current => current.EndsWith(name, StringComparison.OrdinalIgnoreCase));

        return actualName is { } ? Assembly.GetManifestResourceStream(actualName) : default;
    }
}

public class EmbeddedResourceService<T> : EmbeddedResourceService
{
    public EmbeddedResourceService() : base(typeof(T).Assembly) { }
}

Dependency injection

There's no way it would work without dependency injection in the long run so of course there is one here too where the pipelines is resolved from the container by the Request helper extension InvokeAsync. This is where the request's <T> parameter is used to cast the result to what is expected.

public static class RequestExtensions
{
    public static async Task<T> InvokeAsync<T>(this IRequest<T> request, IComponentContext components)
    {
        var node =
            components.ResolveOptionalNamed<Service.PipelineBuilder>(request.Tag()) is { } builder
                ? builder.Build()
                : throw DynamicException.Create("PipelineNotFound", $"There is no pipeline to invoke {request.GetType().ToPrettyString()}");
        return
            await node.InvokeAsync(request) is T result
                ? result
                : throw DynamicException.Create("Request", $"{request.GetType().ToPrettyString()} did not return any result.");
    }
}

Utilities

The first version used interfaces to access properties that middleware nodes can modify or read. I found this was too inflexible so I created the PropertyAccessor to be able to select any property and a Condition with ConditionBag to be fully flexible when setting rules for caching-lifetimes

PropertyAccesor

public interface IPropertyAccessor<TValue>
{
    Func<object, TValue> GetValue { get; }

    Action<object, TValue> SetValue { get; }
}

public record PropertyAccessor<TValue>(Func<object, TValue> GetValue, Action<object, TValue> SetValue) : IPropertyAccessor<TValue>;

public static class PropertyService
{
    private static readonly ConcurrentDictionary<(Type, string), object> Cache = new();

    public abstract class For<TSource> where TSource : IRequest
    {
        public static IPropertyAccessor<TValue> Select<TValue>(Expression<Func<TSource, TValue>> expression)
        {
            if (expression.Body is not MemberExpression memberExpression)
            {
                throw new ArgumentException($"Expression must be a {nameof(MemberExpression)}");
            }

            return (IPropertyAccessor<TValue>)Cache.GetOrAdd((typeof(TSource), memberExpression.Member.Name), _ =>
            {
                var targetParameter = Expression.Parameter(typeof(object), "target");
                var valueParameter = Expression.Parameter(typeof(TValue), "value");

                var casted = ParameterConverter<TSource>.Rewrite(memberExpression, targetParameter);

                // ((T)target).Property
                var getter =
                    Expression.Lambda<Func<object, TValue>>(
                        casted,
                        targetParameter
                    ).Compile();

                // ((T)target).Property = value
                var setter =
                    Expression.Lambda<Action<object, TValue>>(
                        Expression.Assign(casted, valueParameter),
                        targetParameter, valueParameter
                    ).Compile();

                return new PropertyAccessor<TValue>(getter, setter);
            });
        }
    }
}

Condition

public interface ICondition
{
    Func<object, bool> Evaluate { get; }
}

public record Condition(Func<object, bool> Evaluate) : ICondition
{
    private static readonly ConcurrentDictionary<(Type, string), object> Cache = new();

    public abstract class For<TSource> where TSource : IRequest
    {
        public static ICondition When(Expression<Func<TSource, bool>> expression, string tag)
        {
            return (ICondition)Cache.GetOrAdd((typeof(TSource), tag), _ =>
            {
                var targetParameter = Expression.Parameter(typeof(object), "target");

                var casted = ParameterConverter<TSource>.Rewrite(expression.Body, targetParameter);

                // ((T)target)() -> bool
                var evaluate =
                    Expression.Lambda<Func<object, bool>>(
                        casted,
                        targetParameter
                    ).Compile();

                return new Condition(evaluate);
            });
        }
    }
}

public record ConditionBag<T>(Func<object, bool> Evaluate, Func<T> GetValue) : ICondition;

public static class ConditionBagFactory
{
    public static ConditionBag<T> Then<T>(this ICondition condition, T value)
    {
        return new ConditionBag<T>(condition.Evaluate, () => value);
    }
}

---

What do you think?